Add Control Flow extension

README.md

@@ -49,7 +49,6 @@ Only a curated selection of the extensive [GeoServer extensions](http://geoserve
 * **Catalog and Configuration**:
     * PGConfig
     * JDBC `jdbcconfig` and `jdbcstore` (deprecated)
-    * Optimized Catalog Data-Directory loader
 * **Security**:
     * GeoServer ACL
     * JDBC Security
@@ -75,6 +74,7 @@ Only a curated selection of the extensive [GeoServer extensions](http://geoserve
     * Azure Blob Storage
     * Google Cloud Storage Blob Storage
 * **Miscellaneous**:
+    * [Control flow](https://docs.geoserver.org/main/en/user/extensions/controlflow/index.html)
     * Importer
     * Resource Browser Tool
     * International Astronomical Union CRS authority

compose/.env

@@ -8,7 +8,8 @@ GS_USER="1000:1000"
 
 # logging profile, either "default" or "json-logs"
 #LOGGING_PROFILE=json-logs
-LOGGING_PROFILE=default
+#LOGGING_PROFILE=default
+LOGGING_PROFILE=logging_debug_controlflow
 GEOSERVER_DEFAULT_PROFILES="${LOGGING_PROFILE},acl"
 
 GATEWAY_PORT=9090
@@ -24,4 +25,4 @@ CONFIG_SERVER_DEFAULT_PROFILES=${LOGGING_PROFILE},native,standalone
 
 JAVA_OPTS_DEFAULT=-XshowSettings:system -Dlogging.config=file:/etc/geoserver/logback-spring.xml -Xlog:cds
 
-JAVA_OPTS_GEOSERVER=$JAVA_OPTS_DEFAULT
+JAVA_OPTS_GEOSERVER=$JAVA_OPTS_DEFAULT -Dcontrol-flow=true

compose/compose.yml

@@ -63,7 +63,7 @@ services:
     deploy:
       resources:
         limits:
-          cpus: '4.0'
+          cpus: '2.0'
           memory: 2G
 
   # Spring Cloud Config service, provides centralized configuration to all
@@ -140,8 +140,8 @@ services:
     deploy:
       resources:
         limits:
-          cpus: '4.0'
-          memory: 12G
+          cpus: '2.0'
+          memory: 2G
 
   wcs:
     extends:
@@ -166,7 +166,7 @@ services:
     deploy:
       resources:
         limits:
-          cpus: '4.0'
+          cpus: '2.0'
           memory: 2G
 
   restconfig:

config

@@ -1 +1 @@
-Subproject commit 2ef35f94af3c160afe0589b3851c8839be63bdc9
+Subproject commit 51eecaffca8d0b42f36ad2c66785ed0978890720

pom.xml

@@ -161,6 +161,11 @@
         <artifactId>gs-cloud-extension-ogcapi-features</artifactId>
         <version>${project.version}</version>
       </dependency>
+      <dependency>
+        <groupId>org.geoserver.cloud.extensions</groupId>
+        <artifactId>gs-cloud-extension-control-flow</artifactId>
+        <version>${project.version}</version>
+      </dependency>
       <dependency>
         <groupId>org.geoserver.cloud</groupId>
         <artifactId>gs-cloud-spring-boot3-starter</artifactId>

src/extensions/control-flow/README.md

@@ -0,0 +1,159 @@
+# GeoServer Control Flow Extension
+
+This module integrates the GeoServer Control Flow extension with GeoServer Cloud.
+
+## Overview
+
+The Control Flow extension allows administrators to control and throttle requests to manage server resources effectively. It helps with:
+
+- **Performance**: Achieve optimal throughput by limiting concurrent requests to match available CPU cores
+- **Resource Control**: Prevent OutOfMemoryErrors by controlling the number of parallel requests
+- **Fairness**: Prevent a single user from overwhelming the server, ensuring equitable resource distribution
+
+The control flow method queues excess requests rather than rejecting them, though it can be configured to reject requests that wait too long in the queue.
+
+## Configuration
+
+The extension is **enabled by default**. Configuration can be done in two ways:
+
+### 1. Externalized Configuration (Default)
+
+The recommended approach for GeoServer Cloud uses Spring Boot configuration properties with SpEL expression support:
+
+```yaml
+geoserver:
+  extension:
+    control-flow:
+      enabled: true  # Enable/disable the extension (default: true)
+      use-properties-file: false  # Use externalized config (default: false)
+      properties:
+        '[timeout]': 10  # Request timeout in seconds
+        '[ows.global]': "${cpu.cores} * 2"  # Global OWS request limit
+        '[ows.wms]': "${cpu.cores} * 4"  # WMS service limit
+        '[ows.wms.getmap]': "${cpu.cores} * 2"  # GetMap request limit
+```
+
+The default configuration is provided in `config/geoserver_control_flow.yml`.
+
+### 2. Data Directory Configuration
+
+To use the traditional `control-flow.properties` file in the GeoServer data directory:
+
+```yaml
+geoserver:
+  extension:
+    control-flow:
+      enabled: true
+      use-properties-file: true
+```
+
+## Key Features
+
+### Dynamic Configuration with SpEL
+
+The externalized configuration supports Spring Expression Language (SpEL) for dynamic limits based on allocated CPU cores:
+
+```yaml
+properties:
+  '[ows.global]': "${cpu.cores} * 2"  # Resolves to 2x the number of cores
+```
+
+The `cpu.cores` property is automatically available and reflects the container's allocated CPU resources.
+
+### Request Control Rules
+
+Control can be applied at different granularity levels:
+
+```yaml
+# Global OWS limit
+'[ows.global]': 10
+
+# Per-service limit
+'[ows.wms]': 8
+'[ows.wfs]': 6
+
+# Per-request type
+'[ows.wms.getmap]': 4
+'[ows.wps.execute]': 2
+
+# Per-output format
+'[ows.wfs.getfeature.application/msexcel]': 2
+
+# GeoWebCache services (WMS-C, TMS, WMTS)
+'[ows.gwc]': 16
+```
+
+### User-Based Concurrency Control
+
+Limit concurrent requests per user or IP address:
+
+```yaml
+# Cookie-based user identification
+'[user]': 3
+
+# IP-based identification
+'[ip]': 6
+
+# Specific IP address
+'[ip.10.0.0.1]': 10
+
+# IP blacklist
+'[ip.blacklist]': "192.168.0.7, 192.168.0.8"
+```
+
+### Rate Control
+
+Limit requests per time unit:
+
+```yaml
+# Rate limiting syntax: <requests>/<unit>[;<delay>s]
+# Units: s (second), m (minute), h (hour), d (day)
+'[user.ows.wms.getmap]': "30/s"
+'[user.ows.wps.execute]': "1000/d;30s"
+```
+
+## Dependencies
+
+This extension requires the following GeoServer dependency:
+
+- `gs-control-flow`
+
+## Implementation Details
+
+### Key Classes
+
+- `ControlFlowAutoConfiguration`: Main auto-configuration class
+- `ControlFlowConfigurationProperties`: Configuration properties with SpEL expression support
+- `PropertiesControlFlowConfigurator`: Configurator for externalized properties
+- `ExpressionEvaluator`: Evaluates SpEL expressions and resolves placeholders
+- `ConditionalOnControlFlow`: Composite conditional annotation for enabling the extension
+
+### Configuration Modes
+
+The extension operates in two mutually exclusive modes:
+
+1. **UsingExternalizedConfiguration** (default): Uses `geoserver.extension.control-flow.properties` configuration with SpEL support
+2. **UsingDataDirectoryConfiguration**: Uses traditional `control-flow.properties` file from the data directory
+
+### Beans Registered
+
+- `ControlFlowCallback`: Dispatcher callback that enforces flow control rules
+- `ControlFlowConfigurator`: Reads and parses configuration
+- `FlowControllerProvider`: Provides flow controllers based on configuration
+- `IpBlacklistFilter`: Filters requests from blacklisted IP addresses
+- `ControlModuleStatus`: Reports extension status
+
+## Environment Variable Override
+
+The extension supports a shorthand environment variable for quick enable/disable:
+
+```bash
+export CONTROL_FLOW=false
+```
+
+This works through the property placeholder: `${control-flow:true}`
+
+## Related Documentation
+
+- [GeoServer Control Flow User Guide](https://docs.geoserver.org/main/en/user/extensions/controlflow/index.html)
+- Default configuration: `config/geoserver_control_flow.yml`

src/extensions/control-flow/pom.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.geoserver.cloud.extensions</groupId>
+    <artifactId>gs-cloud-extensions</artifactId>
+    <version>${revision}</version>
+  </parent>
+  <artifactId>gs-cloud-extension-control-flow</artifactId>
+  <packaging>jar</packaging>
+  <description>GeoServer Control-flow extension</description>
+  <dependencies>
+    <dependency>
+      <groupId>org.geoserver.cloud.extensions</groupId>
+      <artifactId>gs-cloud-extensions-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.geoserver.extension</groupId>
+      <artifactId>gs-control-flow</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.geoserver</groupId>
+      <artifactId>gs-main</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>javax.servlet</groupId>
+      <artifactId>javax.servlet-api</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.geoserver</groupId>
+      <artifactId>gs-main</artifactId>
+      <version>${gs.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-configuration-processor</artifactId>
+      <optional>true</optional>
+    </dependency>
+  </dependencies>
+</project>

src/extensions/control-flow/src/main/java/org/geoserver/cloud/autoconfigure/extensions/controlflow/ConditionalOnControlFlow.java

@@ -0,0 +1,21 @@
+/* (c) 2025 Open Source Geospatial Foundation - all rights reserved
+ * This code is licensed under the GPL 2.0 license, available at the root
+ * application directory.
+ */
+
+package org.geoserver.cloud.autoconfigure.extensions.controlflow;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD, ElementType.TYPE})
+@Documented
+@ConditionalOnClass(name = "org.geoserver.flow.ControlFlowConfigurator")
+@ConditionalOnProperty(name = ControlFlowConfigurationProperties.ENABLED, havingValue = "true", matchIfMissing = true)
+@interface ConditionalOnControlFlow {}

src/extensions/control-flow/src/main/java/org/geoserver/cloud/autoconfigure/extensions/controlflow/ControlFlowAppContextInitializer.java

@@ -0,0 +1,20 @@
+/* (c) 2025 Open Source Geospatial Foundation - all rights reserved
+ * This code is licensed under the GPL 2.0 license, available at the root
+ * application directory.
+ */
+package org.geoserver.cloud.autoconfigure.extensions.controlflow;
+
+import org.springframework.context.ApplicationContextInitializer;
+import org.springframework.context.ConfigurableApplicationContext;
+
+public class ControlFlowAppContextInitializer implements ApplicationContextInitializer<ConfigurableApplicationContext> {
+
+    @Override
+    public void initialize(ConfigurableApplicationContext applicationContext) {
+        String cores = System.getProperty("cpu.cores");
+        if (null == cores) {
+            cores = "" + Runtime.getRuntime().availableProcessors();
+            System.setProperty("cpu.cores", cores);
+        }
+    }
+}

src/extensions/control-flow/src/main/java/org/geoserver/cloud/autoconfigure/extensions/controlflow/ControlFlowAutoConfiguration.java

@@ -0,0 +1,112 @@
+/* (c) 2025 Open Source Geospatial Foundation - all rights reserved
+ * This code is licensed under the GPL 2.0 license, available at the root
+ * application directory.
+ */
+package org.geoserver.cloud.autoconfigure.extensions.controlflow;
+
+import static org.geoserver.cloud.autoconfigure.extensions.controlflow.ControlFlowConfigurationProperties.USE_PROPERTIES_FILE;
+
+import java.util.Optional;
+import java.util.Properties;
+import org.geoserver.flow.ControlFlowCallback;
+import org.geoserver.flow.ControlFlowConfigurator;
+import org.geoserver.flow.ControlModuleStatus;
+import org.geoserver.flow.DefaultFlowControllerProvider;
+import org.geoserver.flow.FlowControllerProvider;
+import org.geoserver.flow.config.DefaultControlFlowConfigurator;
+import org.geoserver.flow.controller.IpBlacklistFilter;
+import org.geoserver.platform.GeoServerResourceLoader;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
+@AutoConfiguration
+@Import({
+    ControlFlowAutoConfiguration.Enabled.class,
+    ControlFlowAutoConfiguration.UsingDataDirectoryConfiguration.class,
+    ControlFlowAutoConfiguration.UsingExternalizedConfiguration.class
+})
+@EnableConfigurationProperties(ControlFlowConfigurationProperties.class)
+public class ControlFlowAutoConfiguration {
+
+    @Bean
+    ControlModuleStatus controlExtension(
+            ControlFlowConfigurationProperties config, Optional<ControlFlowCallback> callback) {
+
+        ControlModuleStatus controlExtension = new ControlModuleStatus();
+        controlExtension.setComponent("gs-control-flow");
+        controlExtension.setEnabled(config.isEnabled() && callback.isPresent());
+        controlExtension.setAvailable(callback.isPresent());
+        return controlExtension;
+    }
+
+    /**
+     * Sets up {@link ControlFlowConfigurator} and {@link FlowControllerProvider} when {@code geoserver.extension.control-flow.use-properties-file=true}
+     */
+    @Configuration(proxyBeanMethods = false)
+    @ConditionalOnControlFlow
+    @ConditionalOnProperty(name = USE_PROPERTIES_FILE, havingValue = "true", matchIfMissing = false)
+    static class UsingDataDirectoryConfiguration {
+
+        /**
+         * Parameter {@code loader} added because {@link DefaultControlFlowConfigurator} calls {@code GeoServerExtensions.bean(GeoServerResourceLoader.class)}
+         */
+        @Bean
+        ControlFlowConfigurator dataDirectoryPropertiesFileControlFlowConfigurator(GeoServerResourceLoader loader) {
+            return new DefaultControlFlowConfigurator();
+        }
+
+        @Bean
+        FlowControllerProvider defaultFlowControllerProvider(ControlFlowConfigurator configurator) {
+            return new DefaultFlowControllerProvider(configurator);
+        }
+    }
+
+    /**
+     * Sets up {@link ControlFlowConfigurator} and {@link FlowControllerProvider} when {@code geoserver.extension.control-flow.use-properties-file=false} (default)
+     */
+    @Configuration(proxyBeanMethods = false)
+    @ConditionalOnControlFlow
+    @ConditionalOnProperty(name = USE_PROPERTIES_FILE, havingValue = "false", matchIfMissing = true)
+    static class UsingExternalizedConfiguration {
+
+        @Bean
+        PropertiesControlFlowConfigurator externalizedControlFlowConfigurator(
+                ControlFlowConfigurationProperties config) {
+            return new PropertiesControlFlowConfigurator(config.resolvedProperties());
+        }
+
+        @Bean
+        FlowControllerProvider defaultFlowControllerProvider(PropertiesControlFlowConfigurator configurator) {
+            DefaultFlowControllerProvider provider = new DefaultFlowControllerProvider(configurator);
+            configurator.setStale(false);
+            return provider;
+        }
+    }
+
+    // from applicationContext.xml:
+    @Configuration(proxyBeanMethods = false)
+    @ConditionalOnControlFlow
+    static class Enabled {
+        /**
+         * Parameters {@code provider} and {@code configurator} added to ensure they're
+         * created before
+         * {@link ControlFlowCallback#setApplicationContext(org.springframework.context.ApplicationContext)}
+         * tries to register them itself.
+         */
+        @Bean
+        ControlFlowCallback controlFlowCallback(FlowControllerProvider provider, ControlFlowConfigurator configurator) {
+            return new ControlFlowCallback();
+        }
+
+        @Bean
+        @ConditionalOnControlFlow
+        IpBlacklistFilter ipBlacklistFilter(ControlFlowConfigurationProperties config) {
+            Properties properties = config.resolvedProperties();
+            return new IpBlacklistFilter(properties);
+        }
+    }
+}

src/extensions/control-flow/src/main/java/org/geoserver/cloud/autoconfigure/extensions/controlflow/ControlFlowConfigurationProperties.java

@@ -0,0 +1,75 @@
+/* (c) 2025 Open Source Geospatial Foundation - all rights reserved
+ * This code is licensed under the GPL 2.0 license, available at the root
+ * application directory.
+ */
+package org.geoserver.cloud.autoconfigure.extensions.controlflow;
+
+import java.util.Properties;
+import lombok.Data;
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.core.env.Environment;
+
+@Data
+@ConfigurationProperties(prefix = "geoserver.extension.control-flow")
+@Slf4j(topic = "org.geoserver.cloud.autoconfigure.extensions.controlflow")
+class ControlFlowConfigurationProperties {
+
+    static final String ENABLED = "geoserver.extension.control-flow.enabled";
+    static final String USE_PROPERTIES_FILE = "geoserver.extension.control-flow.use-properties-file";
+
+    private final transient @NonNull ExpressionEvaluator evaluator;
+    private transient Properties resolved;
+
+    /**
+     * Whether to enable the control-flow extension
+     */
+    private boolean enabled = true;
+
+    /**
+     * Whether to use the default control-flow.properties file in the data directory
+     * for configuration
+     */
+    private boolean usePropertiesFile = false;
+
+    /**
+     * key/value pairs of control flow configuration properties. Unused if
+     * geoserver.extension.control-flow.use-properties-file is true
+     */
+    private Properties properties = new Properties();
+
+    ControlFlowConfigurationProperties(Environment environment) {
+        this.evaluator = new ExpressionEvaluator(environment);
+    }
+
+    public Properties resolvedProperties() {
+        if (resolved == null) {
+            resolved = new Properties();
+            for (String name : properties.stringPropertyNames()) {
+                String value = properties.getProperty(name);
+                String resolvedValue = resolve(value);
+                resolved.setProperty(name, resolvedValue);
+            }
+        }
+        return resolved;
+    }
+
+    private String resolve(final String value) {
+        String resolved = evaluator.resolvePlaceholders(value);
+        try {
+            resolved = evaluator.evaluateExpressions(resolved);
+        } catch (Exception e) {
+            log.warn(
+                    """
+                     Error evaluating SpEL expressions in '{}', returning '{}'. \
+                     This is ok if you're not trying to use an SpEL expression to perform arithmetic operations. \
+                     Error message: {}
+                     """,
+                    value,
+                    resolved,
+                    e.getMessage());
+        }
+        return resolved;
+    }
+}

src/extensions/control-flow/src/main/java/org/geoserver/cloud/autoconfigure/extensions/controlflow/ExpressionEvaluator.java

@@ -0,0 +1,34 @@
+/* (c) 2025 Open Source Geospatial Foundation - all rights reserved
+ * This code is licensed under the GPL 2.0 license, available at the root
+ * application directory.
+ */
+package org.geoserver.cloud.autoconfigure.extensions.controlflow;
+
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import org.springframework.core.env.Environment;
+import org.springframework.expression.Expression;
+import org.springframework.expression.ExpressionParser;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+
+@RequiredArgsConstructor
+class ExpressionEvaluator {
+
+    private final @NonNull Environment environment;
+
+    private final ExpressionParser parser = new SpelExpressionParser();
+
+    // Use a standard context, we don't need access to Spring beans within the SpEL
+    // itself
+    private final StandardEvaluationContext context = new StandardEvaluationContext();
+
+    public String resolvePlaceholders(String expressionString) {
+        return environment.resolvePlaceholders(expressionString);
+    }
+
+    public String evaluateExpressions(String expressionString) {
+        Expression exp = parser.parseExpression(expressionString);
+        return exp.getValue(context, String.class);
+    }
+}

src/extensions/control-flow/src/main/java/org/geoserver/cloud/autoconfigure/extensions/controlflow/PropertiesControlFlowConfigurator.java

@@ -0,0 +1,252 @@
+/* (c) 2025 Open Source Geospatial Foundation - all rights reserved
+ * This code is licensed under the GPL 2.0 license, available at the root
+ * application directory.
+ */
+package org.geoserver.cloud.autoconfigure.extensions.controlflow;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.StringTokenizer;
+import java.util.logging.Logger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.geoserver.flow.ControlFlowConfigurator;
+import org.geoserver.flow.FlowController;
+import org.geoserver.flow.config.Intervals;
+import org.geoserver.flow.controller.BasicOWSController;
+import org.geoserver.flow.controller.CookieKeyGenerator;
+import org.geoserver.flow.controller.GlobalFlowController;
+import org.geoserver.flow.controller.HttpHeaderPriorityProvider;
+import org.geoserver.flow.controller.IpFlowController;
+import org.geoserver.flow.controller.IpKeyGenerator;
+import org.geoserver.flow.controller.KeyGenerator;
+import org.geoserver.flow.controller.OWSRequestMatcher;
+import org.geoserver.flow.controller.PriorityProvider;
+import org.geoserver.flow.controller.PriorityThreadBlocker;
+import org.geoserver.flow.controller.RateFlowController;
+import org.geoserver.flow.controller.SimpleThreadBlocker;
+import org.geoserver.flow.controller.SingleIpFlowController;
+import org.geoserver.flow.controller.ThreadBlocker;
+import org.geoserver.flow.controller.UserConcurrentFlowController;
+import org.geotools.util.logging.Logging;
+
+/**
+ * Based on {@link org.geoserver.flow.config.DefaultControlFlowConfigurator}, would require a bit of refactoring to allow extending it
+ * @author Andrea Aime - OpenGeo
+ * @author Juan Marin, OpenGeo
+ */
+class PropertiesControlFlowConfigurator implements ControlFlowConfigurator {
+
+    static final Pattern RATE_PATTERN = Pattern.compile("(\\d+)/([smhd])(;(\\d+)s)?");
+
+    static final Logger LOGGER = Logging.getLogger(PropertiesControlFlowConfigurator.class);
+
+    private Properties properties;
+
+    private boolean stale = true;
+
+    public PropertiesControlFlowConfigurator(Properties properties) {
+        this.properties = properties;
+    }
+
+    private Properties getProperties() {
+        return properties;
+    }
+
+    @Override
+    public boolean isStale() {
+        return stale;
+    }
+
+    public void setStale(boolean b) {
+        this.stale = b;
+    }
+
+    @Override
+    public long getTimeout() {
+        return timeout;
+    }
+
+    /**
+     * Factors out the code to build a rate flow controller
+     *
+     * @author Andrea Aime - GeoSolutions
+     */
+    abstract static class RateControllerBuilder {
+        public FlowController build(String[] keys, String value) {
+            Matcher matcher = RATE_PATTERN.matcher(value);
+            if (!matcher.matches()) {
+                LOGGER.severe("Rate limiting rule values should be expressed as <rate</<unit>[;<delay>s], "
+                        + "where unit can be s, m, h or d. This one is invalid: "
+                        + value);
+                return null;
+            }
+            int rate = Integer.parseInt(matcher.group(1));
+            long interval = Intervals.valueOf(matcher.group(2)).getDuration();
+            int delay = 0;
+            String userDelay = matcher.group(4);
+            if (userDelay != null) {
+                delay = Integer.parseInt(userDelay) * 1000;
+            }
+
+            String service = keys.length >= 3 ? keys[2] : null;
+            String request = keys.length >= 4 ? keys[3] : null;
+            String format = keys.length >= 5 ? keys[4] : null;
+            OWSRequestMatcher requestMatcher = new OWSRequestMatcher(service, request, format);
+            KeyGenerator keyGenerator = buildKeyGenerator(keys, value);
+            return new RateFlowController(requestMatcher, rate, interval, delay, keyGenerator);
+        }
+
+        protected abstract KeyGenerator buildKeyGenerator(String[] keys, String value);
+    }
+
+    long timeout = -1;
+
+    @Override
+    public List<FlowController> buildFlowControllers() throws Exception {
+        timeout = -1;
+
+        Properties p = getProperties();
+        List<FlowController> newControllers = new ArrayList<>();
+        PriorityProvider priorityProvider = getPriorityProvider(p);
+
+        for (Object okey : p.keySet()) {
+            String key = ((String) okey).trim();
+            String value = (String) p.get(okey);
+
+            String[] keys = key.split("\\s*\\.\\s*");
+
+            int queueSize = 0;
+            StringTokenizer tokenizer = new StringTokenizer(value, ",");
+            try {
+                // some properties are not integers
+                if ("ip.blacklist".equals(key) || "ip.whitelist".equals(key) || "ows.priority.http".equals(key)) {
+                    continue;
+                } else {
+                    if (!key.startsWith("user.ows") && !key.startsWith("ip.ows")) {
+                        if (tokenizer.countTokens() == 1) {
+                            queueSize = Integer.parseInt(value);
+                        } else {
+                            queueSize = Integer.parseInt(tokenizer.nextToken());
+                        }
+                    }
+                }
+            } catch (NumberFormatException e) {
+                LOGGER.severe(
+                        "Rules should be assigned just a queue size, instead " + key + " is associated to " + value);
+                continue;
+            }
+
+            FlowController controller = null;
+            if ("timeout".equalsIgnoreCase(key)) {
+                timeout = queueSize * 1000;
+                continue;
+            }
+            if ("ows.global".equalsIgnoreCase(key)) {
+                controller = new GlobalFlowController(queueSize, buildBlocker(queueSize, priorityProvider));
+            } else if ("ows".equals(keys[0])) {
+                // todo: check, if possible, if the service, method and output format actually exist
+                ThreadBlocker threadBlocker = buildBlocker(queueSize, priorityProvider);
+                if (keys.length >= 4) {
+                    controller = new BasicOWSController(keys[1], keys[2], keys[3], queueSize, threadBlocker);
+                } else if (keys.length == 3) {
+                    controller = new BasicOWSController(keys[1], keys[2], queueSize, threadBlocker);
+                } else if (keys.length == 2) {
+                    controller = new BasicOWSController(keys[1], queueSize, threadBlocker);
+                }
+            } else if ("user".equals(keys[0])) {
+                if (keys.length == 1) {
+                    controller = new UserConcurrentFlowController(queueSize);
+                } else if ("ows".equals(keys[1])) {
+                    controller = new RateControllerBuilder() {
+
+                        @Override
+                        protected KeyGenerator buildKeyGenerator(String[] keys, String value) {
+                            return new CookieKeyGenerator();
+                        }
+                    }.build(keys, value);
+                }
+            } else if ("ip".equals(keys[0])) {
+                if (keys.length == 1) {
+                    controller = new IpFlowController(queueSize);
+                } else if (keys.length > 1 && "ows".equals(keys[1])) {
+                    controller = new RateControllerBuilder() {
+
+                        @Override
+                        protected KeyGenerator buildKeyGenerator(String[] keys, String value) {
+                            return new IpKeyGenerator();
+                        }
+                    }.build(keys, value);
+                } else if (keys.length > 1) {
+                    if (!"blacklist".equals(keys[1]) && !"whitelist".equals(keys[1])) {
+                        String ip = key.substring("ip.".length());
+                        controller = new SingleIpFlowController(queueSize, ip);
+                    }
+                }
+            }
+
+            if (controller == null) {
+                LOGGER.severe("Could not parse control-flow rule: '" + okey + "=" + value);
+            } else {
+                LOGGER.info("Loaded control-flow rule: " + key + "=" + value);
+                newControllers.add(controller);
+            }
+        }
+
+        return newControllers;
+    }
+
+    /**
+     * Parses the configuration for priority providers
+     *
+     * @param p the configuration properties
+     * @return A {@link PriorityProvider} or null if no (valid) configuration was found
+     */
+    private PriorityProvider getPriorityProvider(Properties p) {
+        for (Object okey : p.keySet()) {
+            String key = ((String) okey).trim();
+            String value = (String) p.get(okey);
+
+            // is it a priority specification?
+            if ("ows.priority.http".equals(key)) {
+                String error = "";
+                try {
+                    String[] splitValue = value.trim().split("\\s*,\\s*");
+                    if (splitValue.length == 2 && splitValue[0].length() > 0) {
+                        String httpHeaderName = splitValue[0];
+                        int defaultPriority = Integer.parseInt(splitValue[1]);
+
+                        LOGGER.info("Found OWS priority specification " + key + "=" + value);
+                        return new HttpHeaderPriorityProvider(httpHeaderName, defaultPriority);
+                    }
+                } catch (NumberFormatException e) {
+                    error = " " + e.getMessage();
+                }
+
+                LOGGER.severe("Unexpected priority specification found '"
+                        + value
+                        + "', "
+                        + "the expected format is headerName,defaultPriorityValue."
+                        + error);
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Builds a {@link ThreadBlocker} based on a queue size and a prority provider
+     *
+     * @param queueSize The count of concurrent requests allowed to run
+     * @param priorityProvider The priority provider (if not null, a
+     *     {@link org.geoserver.flow.controller.PriorityThreadBlocker} will be built
+     * @return a {@link ThreadBlocker}
+     */
+    private ThreadBlocker buildBlocker(int queueSize, PriorityProvider priorityProvider) {
+        if (priorityProvider != null) {
+            return new PriorityThreadBlocker(queueSize, priorityProvider);
+        } else {
+            return new SimpleThreadBlocker(queueSize);
+        }
+    }
+}

src/extensions/control-flow/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,6 @@
+# Initializers
+org.springframework.context.ApplicationContextInitializer=\
+org.geoserver.cloud.autoconfigure.extensions.controlflow.ControlFlowAppContextInitializer
+
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.geoserver.cloud.autoconfigure.extensions.controlflow.ControlFlowAutoConfiguration

src/extensions/control-flow/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+org.geoserver.cloud.autoconfigure.extensions.controlflow.ControlFlowAutoConfiguration

src/extensions/control-flow/src/test/java/org/geoserver/cloud/autoconfigure/extensions/controlflow/ControlFlowAutoConfigurationIT.java

@@ -0,0 +1,62 @@
+/* (c) 2025 Open Source Geospatial Foundation - all rights reserved
+ * This code is licensed under the GPL 2.0 license, available at the root
+ * application directory.
+ */
+package org.geoserver.cloud.autoconfigure.extensions.controlflow;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Properties;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.ApplicationContext;
+
+@SpringBootTest(classes = ControlFlowAutoConfiguration.class)
+@Slf4j
+class ControlFlowAutoConfigurationIT {
+
+    @Autowired
+    ControlFlowConfigurationProperties config;
+
+    @Autowired
+    ApplicationContext context;
+
+    @Test
+    void testControlFlowAppContextInitializer() {
+        String cores = context.getEnvironment().getProperty("cpu.cores");
+        assertThat(cores).isEqualTo(Runtime.getRuntime().availableProcessors() + "");
+    }
+
+    /**
+     * See {@literal src/test/resources/application.yml}
+     */
+    @Test
+    void testResolvedProperties() {
+        Properties props = config.getProperties();
+        Properties resolved = config.resolvedProperties();
+        log.info("control-flow unresolved props: {}", props);
+        log.info("control-flow   resolved props: {}", resolved);
+
+        int cores = Runtime.getRuntime().availableProcessors();
+        String coresTimes2 = String.valueOf(2 * cores);
+        String coresTimes4 = String.valueOf(4 * cores);
+        String halfCores = String.valueOf(cores / 2);
+
+        assertThat(resolved.getProperty("ows.global")).isEqualTo(coresTimes2);
+        assertThat(resolved.getProperty("ows.wms")).isEqualTo(String.valueOf(cores));
+        assertThat(resolved.getProperty("ows.wms.getmap")).isEqualTo(halfCores);
+        assertThat(resolved.getProperty("ows.gwc")).isEqualTo(coresTimes4);
+        assertThat(resolved.getProperty("ows.wfs.getfeature.application/msexcel"))
+                .isEqualTo("2");
+
+        assertThat(resolved.getProperty("timeout")).isEqualTo("10");
+        assertThat(resolved.getProperty("user")).isEqualTo(String.valueOf(cores));
+        assertThat(resolved.getProperty("user.ows.wps.execute")).isEqualTo("1000/d;30s");
+
+        assertThat(resolved.getProperty("ip")).isEqualTo("6");
+        assertThat(resolved.getProperty("ip.10.0.0.1")).isEqualTo(String.valueOf(3 * cores));
+        assertThat(resolved.getProperty("ip.blacklist")).isEqualTo("192.168.0.7, 192.168.0.8");
+    }
+}

src/extensions/control-flow/src/test/java/org/geoserver/cloud/autoconfigure/extensions/controlflow/ControlFlowAutoConfigurationTest.java

@@ -0,0 +1,133 @@
+/* (c) 2025 Open Source Geospatial Foundation - all rights reserved
+ * This code is licensed under the GPL 2.0 license, available at the root
+ * application directory.
+ */
+package org.geoserver.cloud.autoconfigure.extensions.controlflow;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import org.geoserver.flow.ControlFlowCallback;
+import org.geoserver.flow.ControlFlowConfigurator;
+import org.geoserver.flow.ControlModuleStatus;
+import org.geoserver.flow.DefaultFlowControllerProvider;
+import org.geoserver.flow.FlowControllerProvider;
+import org.geoserver.flow.config.DefaultControlFlowConfigurator;
+import org.geoserver.flow.controller.IpBlacklistFilter;
+import org.geoserver.platform.GeoServerExtensionsHelper;
+import org.geoserver.platform.GeoServerResourceLoader;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.FilteredClassLoader;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+class ControlFlowAutoConfigurationTest {
+
+    private ApplicationContextRunner runner = new ApplicationContextRunner()
+            .withInitializer(new ControlFlowAppContextInitializer())
+            .withConfiguration(AutoConfigurations.of(ControlFlowAutoConfiguration.class));
+
+    @Test
+    void defaultEnabled() {
+        runner.run(context -> assertThat(context)
+                .hasNotFailed()
+                .hasSingleBean(ControlFlowConfigurationProperties.class)
+                .hasSingleBean(ControlModuleStatus.class)
+                .getBean(ControlModuleStatus.class)
+                .hasFieldOrPropertyWithValue("enabled", true)
+                .hasFieldOrPropertyWithValue("available", true));
+    }
+
+    @Test
+    void disabled() {
+        runner.withPropertyValues("geoserver.extension.control-flow.enabled: false")
+                .run(context -> assertThat(context)
+                        .hasNotFailed()
+                        .hasSingleBean(ControlFlowConfigurationProperties.class)
+                        .doesNotHaveBean(ControlFlowCallback.class)
+                        .hasSingleBean(ControlModuleStatus.class)
+                        .getBean(ControlModuleStatus.class)
+                        .hasFieldOrPropertyWithValue("enabled", false)
+                        .hasFieldOrPropertyWithValue("available", false));
+    }
+
+    @Test
+    void conditionalOnClass() {
+        runner.withClassLoader(new FilteredClassLoader(org.geoserver.flow.ControlFlowConfigurator.class))
+                .withPropertyValues("geoserver.extension.control-flow.enabled: true")
+                .run(context -> assertThat(context)
+                        .hasNotFailed()
+                        .hasSingleBean(ControlFlowConfigurationProperties.class)
+                        .doesNotHaveBean(ControlFlowCallback.class)
+                        .hasSingleBean(ControlModuleStatus.class)
+                        .getBean(ControlModuleStatus.class)
+                        .hasFieldOrPropertyWithValue("enabled", false)
+                        .hasFieldOrPropertyWithValue("available", false));
+    }
+
+    @Nested
+    class Enabled {
+
+        @Test
+        void requiredBeans() {
+            runner.run(context -> assertThat(context)
+                    .hasNotFailed()
+                    .hasSingleBean(ControlFlowCallback.class)
+                    .hasSingleBean(IpBlacklistFilter.class));
+        }
+    }
+
+    @Nested
+    class UsingExternalizedConfiguration {
+        @Test
+        void requiredBeans() {
+            runner.run(context -> {
+                assertThat(context)
+                        .hasNotFailed()
+                        .hasSingleBean(ControlFlowConfigurator.class)
+                        .getBean(ControlFlowConfigurator.class)
+                        .isInstanceOf(PropertiesControlFlowConfigurator.class)
+                        .hasFieldOrPropertyWithValue("stale", false);
+
+                assertThat(context)
+                        .hasNotFailed()
+                        .hasSingleBean(FlowControllerProvider.class)
+                        .getBean(FlowControllerProvider.class)
+                        .isInstanceOf(DefaultFlowControllerProvider.class);
+            });
+        }
+    }
+
+    @Nested
+    class UsingDataDirectoryConfiguration {
+        @AfterEach
+        void after() {
+            GeoServerExtensionsHelper.clear();
+        }
+
+        @Test
+        void requiredBeans(@TempDir File tmpDir) {
+            GeoServerResourceLoader resourceLoader = new GeoServerResourceLoader(tmpDir);
+            GeoServerExtensionsHelper.singleton("resourceLoader", resourceLoader, GeoServerResourceLoader.class);
+
+            runner.withPropertyValues("geoserver.extension.control-flow.use-properties-file: true")
+                    .withBean(GeoServerResourceLoader.class, () -> resourceLoader)
+                    .run(context -> {
+                        assertThat(context)
+                                .hasNotFailed()
+                                .hasSingleBean(ControlFlowConfigurator.class)
+                                .getBean(ControlFlowConfigurator.class)
+                                .isInstanceOf(DefaultControlFlowConfigurator.class);
+
+                        assertThat(context)
+                                .hasNotFailed()
+                                .hasSingleBean(FlowControllerProvider.class)
+                                .getBean(FlowControllerProvider.class)
+                                .isInstanceOf(DefaultFlowControllerProvider.class);
+                    });
+        }
+    }
+}

src/extensions/control-flow/src/test/resources/application.yml

@@ -0,0 +1,64 @@
+spring:
+  main:
+    banner-mode: off
+logging:
+  level:
+    org:
+      geoserver:
+        flow: debug 
+        '[cloud.autoconfigure.extensions.controlflow]': debug 
+
+geoserver:
+  extension:
+    control-flow:
+      enabled: true
+      # Whether to use the traditional control-flow.properties file in the data directory
+      # for configuration or these externalized configuration properties
+      use-properties-file: false
+      properties:
+        ##############################################
+        ### TIMEOUT
+        ### Number of seconds a request can stay queued waiting for execution. If the request does not enter execution
+        ### before the timeout expires it will be rejected.
+        '[timeout]': 10
+        ##############################################
+        ### Total OWS request count
+        ### ows.global: Global number of OWS requests executing in parallel
+        '[ows.global]': "${cpu.cores} * 2"
+        ##############################################
+        ### PER REQUEST CONTROL
+        ### per request type control can be demanded using the following syntax: ows.<service>[.<request>[.<outputFormat>]]=<count>
+        ### Where:
+        ### <service> is the OWS service in question (at the time of writing can be wms, wfs, wcs)
+        ### <request>, optional, is the request type. For example, for the wms service it can be GetMap, GetFeatureInfo, DescribeLayer, GetLegendGraphics, GetCapabilities
+        ### <outputFormat>, optional, is the output format of the request. For example, for the wms GetMap request it could be image/png, image/gif and so on.
+        '[ows.wms]': "${cpu.cores}"
+        '[ows.wms.getmap]': "${cpu.cores} / 2"
+        '[ows.wfs.getfeature.application/msexcel]': 2
+        ### GeoWebCache contributes three cached tiles services: WMS-C, TMS, and WMTS. It is also possible to use the
+        ### Control flow module to throttle them, by adding the following rule to the configuration file:
+        '[ows.gwc]': "${cpu.cores} * 4"
+        ##############################################
+        ### PER USER CONCURRENCY CONTROL 
+        ### There are two mechanisms to identify user requests. The first one is cookie based, so it will work fine for
+        ### browsers but not as much for other kinds of clients. The second one is ip based, which works for any type of
+        ### client but that can limit all the users sitting behind the same router
+        ### user: maximum number of requests a single user can execute in parallel.
+        ### ip: maximum number of requests a single ip address can execute in parallel.
+        '[user]': "${cpu.cores}"
+        '[ip]': "6"
+        ### It is also possible to make this a bit more specific and throttle a single ip address instead by using the following:
+        ### ip.<ip_addr>: <count>
+        ### Where <count> is the maximum number of requests the ip specified in <ip_addr> will execute in parallel.
+        '[ip.10.0.0.1]': "3 * ${cpu.cores}"
+        ### To reject requests from a list of ip addresses:
+        '[ip.blacklist]': "192.168.0.7, 192.168.0.8"
+        ##############################################
+        ### PER USER RATE CONTROL
+        ### The rate control rules allow to setup the maximum number of requests per unit of time, based either on a cookie or IP address.
+        ### These rules look as follows (see “Per user concurrency control” for the meaning of “user” and “ip”):
+        ### user.ows[.<service>[.<request>[.<outputFormat>]]]=<requests>/<unit>[;<delay>s]
+        ### ip.ows[.<service>[.<request>[.<outputFormat>]]]=<requests>/<unit>[;<delay>s]
+        '[user.ows.wms.getmap]': "30/s"
+        '[user.ows.wps.execute]': "1000/d;30s"
+        ##############################################

src/extensions/control-flow/src/test/resources/logback-test.xml

@@ -0,0 +1,15 @@
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="WARN">
+        <appender-ref ref="STDOUT"/>
+    </root>
+
+    <logger name="org.springframework.test" level="WARN"/>
+    <logger name="org.springframework.boot.test" level="WARN"/>
+    <logger name="org.springframework.context" level="WARN"/>
+</configuration>

src/extensions/pom.xml

@@ -13,6 +13,7 @@
   <modules>
     <module>core</module>
     <module>app-schema</module>
+    <module>control-flow</module>
     <module>inspire</module>
     <module>security</module>
     <module>input-formats</module>

src/pom.xml

@@ -904,6 +904,12 @@
         <artifactId>gs-web-features</artifactId>
         <version>${gs.version}</version>
       </dependency>
+      <!-- Control-flow dependencies -->
+      <dependency>
+        <groupId>org.geoserver.extension</groupId>
+        <artifactId>gs-control-flow</artifactId>
+        <version>${gs.version}</version>
+      </dependency>
       <!-- GeoTools dependencies -->
       <dependency>
         <groupId>org.geotools</groupId>

src/starters/extensions/pom.xml

@@ -50,5 +50,9 @@
       <groupId>org.geoserver.cloud.extensions</groupId>
       <artifactId>gs-cloud-extension-ogcapi-features</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.geoserver.cloud.extensions</groupId>
+      <artifactId>gs-cloud-extension-control-flow</artifactId>
+    </dependency>
   </dependencies>
 </project>