Gateway Plugins — Extending API Gateway Functionality
In this tutorial, you'll learn about Gateway Plugins. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Gateway plugins extend the functionality of your API gateway with reusable, composable modules for authentication, Rate Limiting, logging, transformation, and more.
What You'll Learn
By the end of this lesson, you will understand the plugin execution lifecycle, configure built-in plugins, chain plugins in the correct order, and manage plugin configuration across environments.
Why It Matters
Plugins provide a modular way to add cross-cutting concerns to your gateway without modifying backend services, enabling rapid addition of new capabilities.
Real-World Use
Durga Antivirus Pro uses gateway plugins for authentication, rate limiting, request logging, response compression, and CORS enforcement, all configured without modifying backend code.
Plugin Execution Lifecycle
flowchart LR
Request-->Plugins[Plugin Chain]
subgraph Plugins
P1[Auth Plugin]
P2[Rate Limit Plugin]
P3[Log Plugin]
P4[Transform Plugin]
end
P1-->P2
P2-->P3
P3-->P4
P4-->Backend[Backend Service]
Backend-->ResponsePlugins[Response Plugins]
ResponsePlugins-->P5[Compression Plugin]
ResponsePlugins-->P6[CORS Plugin]
P5-->P6
P6-->Client
Plugin Architecture
A plugin system with request and response phases.
from typing import Dict, List, Optional, Callable, Any
import time
class GatewayPlugin:
def __init__(self, name: str,
phase: str = "request",
priority: int = 100):
self.name = name
self.phase = phase
self.priority = priority
self.enabled = True
self.config: Dict = {}
def execute(self, request: Dict,
response: Optional[Dict] = None
) -> Dict:
raise NotImplementedError
class PluginManager:
def __init__(self):
self.request_plugins: List[GatewayPlugin] = []
self.response_plugins: List[GatewayPlugin] = []
def register(self, plugin: GatewayPlugin):
if plugin.phase == "request":
self.request_plugins.append(plugin)
self.request_plugins.sort(
key=lambda p: p.priority
)
else:
self.response_plugins.append(plugin)
self.response_plugins.sort(
key=lambda p: p.priority
)
def execute_request_plugins(self,
request: Dict) -> Dict:
context = {"request": request}
for plugin in self.request_plugins:
if plugin.enabled:
context = plugin.execute(context)
return context
def execute_response_plugins(self,
response: Dict) -> Dict:
context = {"response": response}
for plugin in self.response_plugins:
if plugin.enabled:
context = plugin.execute(context)
return context
def disable_plugin(self, name: str):
for plugin in (self.request_plugins +
self.response_plugins):
if plugin.name == name:
plugin.enabled = False
def get_plugin(self, name: str
) -> Optional[GatewayPlugin]:
for plugin in (self.request_plugins +
self.response_plugins):
if plugin.name == name:
return plugin
return None
manager = PluginManager()
Built-in Plugin Examples
Example implementations of common gateway plugins.
class AuthPlugin(GatewayPlugin):
def __init__(self):
super().__init__("authentication", "request", 10)
self.config["public_paths"] = {"/health", "/docs"}
def execute(self, context: Dict) -> Dict:
request = context.get("request", {})
path = request.get("path", "")
if path in self.config["public_paths"]:
return context
auth_header = request.get("headers", {}).get(
"Authorization"
)
if not auth_header:
context["error"] = {
"status": 401,
"body": {"error": "missing_auth"}
}
return context
class RateLimitPlugin(GatewayPlugin):
def __init__(self):
super().__init__("rate_limit", "request", 20)
self.config["max_requests"] = 100
self.config["window_seconds"] = 60
def execute(self, context: Dict) -> Dict:
request = context.get("request", {})
client_ip = request.get("client_ip", "unknown")
has_error = context.get("error")
if not has_error:
context["ratelimit_remaining"] = 99
return context
class LogPlugin(GatewayPlugin):
def __init__(self):
super().__init__("logging", "request", 90)
def execute(self, context: Dict) -> Dict:
request = context.get("request", {})
print(f"Request: {request.get('method')} "
f"{request.get('path')}")
return context
class CompressionPlugin(GatewayPlugin):
def __init__(self):
super().__init__("compression", "response", 10)
def execute(self, context: Dict) -> Dict:
response = context.get("response", {})
response["headers"] = response.get("headers", {})
response["headers"]["Content-Encoding"] = "gzip"
return context
manager.register(AuthPlugin())
manager.register(RateLimitPlugin())
manager.register(LogPlugin())
manager.register(CompressionPlugin())
context = manager.execute_request_plugins({
"method": "GET",
"path": "/api/scan",
"headers": {}
})
print(f"After plugins: error={context.get('error')}")
Plugin Configuration Management
Manage plugin configuration via a central config system.
from typing import Dict, Optional, Any
import json
class PluginConfigManager:
def __init__(self):
self.configs: Dict[str, Dict] = {}
def set_plugin_config(self, plugin_name: str,
config: Dict):
self.configs[plugin_name] = config
def get_plugin_config(self, plugin_name: str
) -> Dict:
return self.configs.get(plugin_name, {})
def merge_config(self, plugin_name: str,
env: str,
base_config: Dict,
env_overrides: Dict) -> Dict:
config = dict(base_config)
overrides = env_overrides.get(env, {})
for key, value in overrides.items():
if key in config:
config[key] = value
return config
def validate_config(self, plugin_name: str,
config: Dict) -> bool:
required_fields = {
"authentication": ["public_paths"],
"rate_limit": ["max_requests", "window_seconds"],
"logging": [],
"compression": [],
}
required = required_fields.get(plugin_name, [])
for field in required:
if field not in config:
print(f"Missing required field: {field}")
return False
return True
def to_json(self, plugin_name: str) -> str:
return json.dumps(
self.get_plugin_config(plugin_name),
indent=2
)
config_mgr = PluginConfigManager()
config_mgr.set_plugin_config("rate_limit", {
"max_requests": 100,
"window_seconds": 60,
"enabled": True
})
valid = config_mgr.validate_config(
"rate_limit",
config_mgr.get_plugin_config("rate_limit")
)
print(f"Config valid: {valid}")
Common Mistakes
Mistake 1: Incorrect Plugin Order
Running authentication after logging logs unauthenticated requests. Run auth first.
Mistake 2: Plugin Execution Timeouts
A slow plugin blocks the entire request pipeline. Set timeouts for plugin execution.
Mistake 3: State Leak Between Plugins
Plugins should not mutate shared state. Each plugin gets its own context namespace.
Mistake 4: No Plugin Isolation
A crashing plugin should not bring down the gateway. Run plugins in isolated contexts.
Mistake 5: Hardcoded Plugin Config
Plugin configuration should be externalized and environment-specific.
Practice Questions
- What is the plugin execution order for request and response phases?
- How do you pass data between plugins in the chain?
- What happens when a plugin fails or throws an exception?
- How do you configure plugins differently per environment?
- What is the difference between a plugin and a middleware?
Challenge
Build a plugin system for the gateway that supports request and response phase plugins with configurable priority ordering, plugin configuration via JSON, error handling that skips a plugin on failure without breaking the chain, and a mechanism to disable specific plugins per route.
FAQ
Mini Project
Build a plugin system for the gateway that includes an authentication plugin (validates JWT, priority 10), a rate limiting plugin (100 req/min per client, priority 20), a logging plugin (structured JSON logs, priority 90), and a compression plugin (gzip responses, priority 10 response), with configuration via a JSON config file.
What's Next
Learn about Plugin Development for creating custom gateway plugins, or explore Gateway Kubernetes for deploying the gateway on Kubernetes.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro