feat: Add complete HSP implementation with integration tests passing
Initial implementation of HTTP Sender Plugin following TDD methodology with hexagonal architecture. All 313 tests passing (0 failures). This commit adds: - Complete domain model and port interfaces - All adapter implementations (HTTP, gRPC, file logging, config) - Application services (data collection, transmission, backpressure) - Comprehensive test suite with 18 integration tests Test fixes applied during implementation: - Fix base64 encoding validation in DataCollectionServiceIntegrationTest - Fix exception type handling in IConfigurationPortTest - Fix CompletionException unwrapping in IHttpPollingPortTest - Fix sequential batching in DataTransmissionServiceIntegrationTest - Add test adapter failure simulation for reconnection tests - Use adapter counters for gRPC verification Files added: - pom.xml with all dependencies (JUnit 5, Mockito, WireMock, gRPC, Jackson) - src/main/java: Domain model, ports, adapters, application services - src/test/java: Unit tests, integration tests, test utilities
This commit is contained in:
Christoph Wagner
@@ -0,0 +1,330 @@
|
||||
{
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"$id": "https://coreshield.siemens.com/schemas/hsp-config-v1.json",
|
||||
"title": "HTTP Sender Plugin Configuration",
|
||||
"description": "Configuration schema for HSP diagnostic data collection system",
|
||||
"type": "object",
|
||||
"required": [
|
||||
"endpoints",
|
||||
"grpc_connection",
|
||||
"buffer"
|
||||
],
|
||||
"properties": {
|
||||
"http_polling": {
|
||||
"type": "object",
|
||||
"description": "HTTP polling configuration",
|
||||
"properties": {
|
||||
"timeout_seconds": {
|
||||
"type": "integer",
|
||||
"description": "HTTP request timeout in seconds",
|
||||
"default": 30,
|
||||
"minimum": 1,
|
||||
"maximum": 300
|
||||
},
|
||||
"retry_attempts": {
|
||||
"type": "integer",
|
||||
"description": "Number of retry attempts for failed requests",
|
||||
"default": 3,
|
||||
"minimum": 0,
|
||||
"maximum": 10
|
||||
},
|
||||
"retry_interval_seconds": {
|
||||
"type": "integer",
|
||||
"description": "Interval between retry attempts in seconds",
|
||||
"default": 5,
|
||||
"minimum": 1,
|
||||
"maximum": 60
|
||||
},
|
||||
"rate_limiting": {
|
||||
"type": "object",
|
||||
"description": "Rate limiting configuration (Phase 1.1)",
|
||||
"properties": {
|
||||
"enabled": {
|
||||
"type": "boolean",
|
||||
"description": "Enable rate limiting for HTTP requests",
|
||||
"default": true
|
||||
},
|
||||
"requests_per_second": {
|
||||
"type": "number",
|
||||
"description": "Maximum requests per second (global)",
|
||||
"default": 10.0,
|
||||
"exclusiveMinimum": 0,
|
||||
"maximum": 1000.0
|
||||
},
|
||||
"per_endpoint": {
|
||||
"type": "boolean",
|
||||
"description": "Apply rate limit per endpoint (true) or globally (false)",
|
||||
"default": true
|
||||
}
|
||||
},
|
||||
"required": ["enabled", "requests_per_second"]
|
||||
},
|
||||
"backpressure": {
|
||||
"type": "object",
|
||||
"description": "Backpressure configuration (Phase 1.2)",
|
||||
"properties": {
|
||||
"enabled": {
|
||||
"type": "boolean",
|
||||
"description": "Enable backpressure mechanism",
|
||||
"default": true
|
||||
},
|
||||
"monitor_interval_ms": {
|
||||
"type": "integer",
|
||||
"description": "Buffer monitoring interval in milliseconds",
|
||||
"default": 100,
|
||||
"minimum": 10,
|
||||
"maximum": 1000
|
||||
},
|
||||
"threshold_percent": {
|
||||
"type": "number",
|
||||
"description": "Buffer usage threshold to trigger backpressure (0-100)",
|
||||
"default": 80.0,
|
||||
"minimum": 0.0,
|
||||
"maximum": 100.0
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"endpoints": {
|
||||
"type": "array",
|
||||
"description": "List of HTTP endpoints to poll",
|
||||
"minItems": 1,
|
||||
"maxItems": 1000,
|
||||
"items": {
|
||||
"type": "object",
|
||||
"required": ["url", "poll_interval_seconds"],
|
||||
"properties": {
|
||||
"url": {
|
||||
"type": "string",
|
||||
"description": "HTTP endpoint URL",
|
||||
"format": "uri",
|
||||
"pattern": "^https?://.+"
|
||||
},
|
||||
"poll_interval_seconds": {
|
||||
"type": "integer",
|
||||
"description": "Polling interval in seconds",
|
||||
"minimum": 1,
|
||||
"maximum": 3600
|
||||
},
|
||||
"enabled": {
|
||||
"type": "boolean",
|
||||
"description": "Enable polling for this endpoint",
|
||||
"default": true
|
||||
},
|
||||
"rate_limit_override": {
|
||||
"type": "number",
|
||||
"description": "Per-endpoint rate limit override (requests per second)",
|
||||
"exclusiveMinimum": 0,
|
||||
"maximum": 1000.0
|
||||
},
|
||||
"priority": {
|
||||
"type": "string",
|
||||
"description": "Endpoint priority (for future use)",
|
||||
"enum": ["low", "normal", "high", "critical"],
|
||||
"default": "normal"
|
||||
},
|
||||
"metadata": {
|
||||
"type": "object",
|
||||
"description": "Additional endpoint metadata",
|
||||
"properties": {
|
||||
"device_id": {
|
||||
"type": "string"
|
||||
},
|
||||
"location": {
|
||||
"type": "string"
|
||||
},
|
||||
"tags": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "string"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"grpc_connection": {
|
||||
"type": "object",
|
||||
"description": "gRPC connection configuration",
|
||||
"required": ["host", "port"],
|
||||
"properties": {
|
||||
"host": {
|
||||
"type": "string",
|
||||
"description": "gRPC server hostname",
|
||||
"minLength": 1
|
||||
},
|
||||
"port": {
|
||||
"type": "integer",
|
||||
"description": "gRPC server port",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
},
|
||||
"receiver_id": {
|
||||
"type": "integer",
|
||||
"description": "Receiver ID for gRPC transmission",
|
||||
"default": 99
|
||||
},
|
||||
"reconnect_interval_seconds": {
|
||||
"type": "integer",
|
||||
"description": "Reconnection interval on failure (seconds)",
|
||||
"default": 5,
|
||||
"minimum": 1,
|
||||
"maximum": 60
|
||||
},
|
||||
"max_reconnect_attempts": {
|
||||
"type": "integer",
|
||||
"description": "Maximum reconnection attempts (0 = infinite)",
|
||||
"default": 0,
|
||||
"minimum": 0,
|
||||
"maximum": 100
|
||||
},
|
||||
"tls": {
|
||||
"type": "object",
|
||||
"description": "TLS configuration (future enhancement)",
|
||||
"properties": {
|
||||
"enabled": {
|
||||
"type": "boolean",
|
||||
"default": false
|
||||
},
|
||||
"cert_path": {
|
||||
"type": "string"
|
||||
},
|
||||
"key_path": {
|
||||
"type": "string"
|
||||
},
|
||||
"ca_cert_path": {
|
||||
"type": "string"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"buffer": {
|
||||
"type": "object",
|
||||
"description": "Circular buffer configuration",
|
||||
"required": ["capacity"],
|
||||
"properties": {
|
||||
"capacity": {
|
||||
"type": "integer",
|
||||
"description": "Buffer capacity (number of messages)",
|
||||
"default": 300,
|
||||
"minimum": 10,
|
||||
"maximum": 10000
|
||||
},
|
||||
"overflow_policy": {
|
||||
"type": "string",
|
||||
"description": "Behavior when buffer is full",
|
||||
"enum": ["discard_oldest", "block", "discard_newest"],
|
||||
"default": "discard_oldest"
|
||||
},
|
||||
"statistics_enabled": {
|
||||
"type": "boolean",
|
||||
"description": "Enable buffer statistics collection",
|
||||
"default": true
|
||||
}
|
||||
}
|
||||
},
|
||||
"transmission": {
|
||||
"type": "object",
|
||||
"description": "Data transmission configuration",
|
||||
"properties": {
|
||||
"batch_size_bytes": {
|
||||
"type": "integer",
|
||||
"description": "Maximum batch size in bytes",
|
||||
"default": 4194304,
|
||||
"minimum": 1024,
|
||||
"maximum": 10485760
|
||||
},
|
||||
"batch_timeout_seconds": {
|
||||
"type": "integer",
|
||||
"description": "Maximum time to accumulate batch (seconds)",
|
||||
"default": 1,
|
||||
"minimum": 1,
|
||||
"maximum": 60
|
||||
}
|
||||
}
|
||||
},
|
||||
"health_check": {
|
||||
"type": "object",
|
||||
"description": "Health check endpoint configuration",
|
||||
"properties": {
|
||||
"enabled": {
|
||||
"type": "boolean",
|
||||
"description": "Enable health check endpoint",
|
||||
"default": true
|
||||
},
|
||||
"port": {
|
||||
"type": "integer",
|
||||
"description": "Health check HTTP server port",
|
||||
"default": 8080,
|
||||
"minimum": 1024,
|
||||
"maximum": 65535
|
||||
},
|
||||
"path": {
|
||||
"type": "string",
|
||||
"description": "Health check endpoint path",
|
||||
"default": "/health",
|
||||
"pattern": "^/.+"
|
||||
}
|
||||
}
|
||||
},
|
||||
"logging": {
|
||||
"type": "object",
|
||||
"description": "Logging configuration",
|
||||
"properties": {
|
||||
"level": {
|
||||
"type": "string",
|
||||
"description": "Log level",
|
||||
"enum": ["TRACE", "DEBUG", "INFO", "WARN", "ERROR"],
|
||||
"default": "INFO"
|
||||
},
|
||||
"file_path": {
|
||||
"type": "string",
|
||||
"description": "Log file path (default: temp directory)",
|
||||
"default": "${java.io.tmpdir}/hsp.log"
|
||||
},
|
||||
"max_file_size_mb": {
|
||||
"type": "integer",
|
||||
"description": "Maximum log file size in MB",
|
||||
"default": 100,
|
||||
"minimum": 1,
|
||||
"maximum": 1000
|
||||
},
|
||||
"max_files": {
|
||||
"type": "integer",
|
||||
"description": "Maximum number of log files to keep",
|
||||
"default": 5,
|
||||
"minimum": 1,
|
||||
"maximum": 100
|
||||
}
|
||||
}
|
||||
},
|
||||
"performance": {
|
||||
"type": "object",
|
||||
"description": "Performance tuning configuration",
|
||||
"properties": {
|
||||
"virtual_threads": {
|
||||
"type": "boolean",
|
||||
"description": "Use Java 25 virtual threads",
|
||||
"default": true
|
||||
},
|
||||
"max_concurrent_polls": {
|
||||
"type": "integer",
|
||||
"description": "Maximum concurrent polling operations",
|
||||
"default": 1000,
|
||||
"minimum": 1,
|
||||
"maximum": 10000
|
||||
},
|
||||
"memory_limit_mb": {
|
||||
"type": "integer",
|
||||
"description": "Memory usage limit in MB",
|
||||
"default": 4096,
|
||||
"minimum": 512,
|
||||
"maximum": 16384
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,254 @@
|
||||
# Rate Limiting Configuration
|
||||
|
||||
## Overview
|
||||
|
||||
The HSP system implements configurable rate limiting for HTTP polling operations to prevent overwhelming endpoint devices and ensure controlled data collection.
|
||||
|
||||
**Requirement**: Req-FR-16 (enhanced)
|
||||
**Phase**: 1.1 - Foundation & Quick Wins
|
||||
**Implementation**: RateLimitedHttpPollingAdapter
|
||||
|
||||
## Configuration Schema
|
||||
|
||||
### JSON Configuration
|
||||
|
||||
Add the following to your `hsp-config.json`:
|
||||
|
||||
```json
|
||||
{
|
||||
"http_polling": {
|
||||
"rate_limiting": {
|
||||
"enabled": true,
|
||||
"requests_per_second": 10.0,
|
||||
"per_endpoint": true
|
||||
}
|
||||
},
|
||||
"endpoints": [
|
||||
{
|
||||
"url": "http://device-1.local/diagnostics",
|
||||
"rate_limit_override": 5.0
|
||||
},
|
||||
{
|
||||
"url": "http://device-2.local/diagnostics",
|
||||
"rate_limit_override": 20.0
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
### Configuration Parameters
|
||||
|
||||
| Parameter | Type | Default | Description | Constraint |
|
||||
|-----------|------|---------|-------------|------------|
|
||||
| `enabled` | boolean | `true` | Enable/disable rate limiting | - |
|
||||
| `requests_per_second` | double | `10.0` | Global rate limit | Must be > 0 |
|
||||
| `per_endpoint` | boolean | `true` | Apply rate limit per endpoint or globally | - |
|
||||
| `rate_limit_override` | double | (optional) | Per-endpoint rate limit override | Must be > 0 |
|
||||
|
||||
## Implementation Details
|
||||
|
||||
### Algorithm
|
||||
|
||||
The implementation uses **Google Guava's RateLimiter**, which implements a token bucket algorithm:
|
||||
|
||||
1. **Token Bucket**: Tokens are added at a constant rate
|
||||
2. **Request Processing**: Each request consumes one token
|
||||
3. **Blocking Behavior**: If no tokens available, request blocks until token is available
|
||||
4. **Smooth Rate**: Distributes requests evenly over time (no bursts)
|
||||
|
||||
### Thread Safety
|
||||
|
||||
- **Thread-Safe**: RateLimiter is thread-safe, allowing concurrent access
|
||||
- **No Locking**: Uses non-blocking algorithms internally
|
||||
- **Fair Distribution**: Requests are served in FIFO order when rate-limited
|
||||
|
||||
### Performance Characteristics
|
||||
|
||||
- **Memory**: O(1) - minimal overhead per instance
|
||||
- **CPU**: O(1) - constant time acquire operation
|
||||
- **Latency**: Average delay = 1 / requests_per_second
|
||||
|
||||
## Usage Examples
|
||||
|
||||
### Example 1: Global Rate Limiting
|
||||
|
||||
```java
|
||||
// Create base HTTP adapter
|
||||
IHttpPollingPort httpAdapter = new HttpPollingAdapter(httpConfig);
|
||||
|
||||
// Wrap with rate limiting (10 requests per second)
|
||||
IHttpPollingPort rateLimited = new RateLimitedHttpPollingAdapter(
|
||||
httpAdapter,
|
||||
10.0 // 10 req/s
|
||||
);
|
||||
|
||||
// Use the rate-limited adapter
|
||||
CompletableFuture<byte[]> data = rateLimited.pollEndpoint("http://device.local/data");
|
||||
```
|
||||
|
||||
### Example 2: Per-Endpoint Rate Limiting
|
||||
|
||||
```java
|
||||
// Different rate limits for different endpoints
|
||||
Map<String, IHttpPollingPort> adapters = new HashMap<>();
|
||||
|
||||
for (EndpointConfig endpoint : config.getEndpoints()) {
|
||||
IHttpPollingPort baseAdapter = new HttpPollingAdapter(endpoint);
|
||||
|
||||
double rateLimit = endpoint.getRateLimitOverride()
|
||||
.orElse(config.getDefaultRateLimit());
|
||||
|
||||
IHttpPollingPort rateLimited = new RateLimitedHttpPollingAdapter(
|
||||
baseAdapter,
|
||||
rateLimit
|
||||
);
|
||||
|
||||
adapters.put(endpoint.getUrl(), rateLimited);
|
||||
}
|
||||
```
|
||||
|
||||
### Example 3: Dynamic Rate Adjustment
|
||||
|
||||
```java
|
||||
// For future enhancement - dynamic rate adjustment
|
||||
public class AdaptiveRateLimiter {
|
||||
private RateLimitedHttpPollingAdapter adapter;
|
||||
|
||||
public void adjustRate(double newRate) {
|
||||
// Would require enhancement to RateLimitedHttpPollingAdapter
|
||||
// to support dynamic rate changes
|
||||
// Currently requires creating new instance
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Testing
|
||||
|
||||
### Test Coverage
|
||||
|
||||
The implementation includes comprehensive tests:
|
||||
|
||||
1. **Initialization Tests**: Valid and invalid configuration
|
||||
2. **Rate Limiting Tests**: Within and exceeding limits
|
||||
3. **Time Window Tests**: Rate limit reset behavior
|
||||
4. **Concurrency Tests**: Thread safety with concurrent requests
|
||||
5. **Burst Traffic Tests**: Handling sudden request spikes
|
||||
6. **Exception Tests**: Error propagation from underlying adapter
|
||||
|
||||
### Running Tests
|
||||
|
||||
```bash
|
||||
# Run unit tests
|
||||
mvn test -Dtest=RateLimitedHttpPollingAdapterTest
|
||||
|
||||
# Generate coverage report
|
||||
mvn test jacoco:report
|
||||
|
||||
# Verify coverage thresholds (95% line, 90% branch)
|
||||
mvn verify
|
||||
```
|
||||
|
||||
## Monitoring
|
||||
|
||||
### Metrics to Monitor
|
||||
|
||||
1. **Rate Limit Wait Time**: Time spent waiting for rate limiter permits
|
||||
2. **Request Throughput**: Actual requests per second achieved
|
||||
3. **Queue Depth**: Number of requests waiting for permits
|
||||
4. **Rate Limit Violations**: Attempts that were throttled
|
||||
|
||||
### Example Monitoring Integration
|
||||
|
||||
```java
|
||||
public class MonitoredRateLimitedAdapter implements IHttpPollingPort {
|
||||
private final RateLimitedHttpPollingAdapter delegate;
|
||||
private final MetricsCollector metrics;
|
||||
|
||||
@Override
|
||||
public CompletableFuture<byte[]> pollEndpoint(String url) {
|
||||
long startTime = System.nanoTime();
|
||||
|
||||
CompletableFuture<byte[]> result = delegate.pollEndpoint(url);
|
||||
|
||||
result.thenRun(() -> {
|
||||
long duration = System.nanoTime() - startTime;
|
||||
metrics.recordRateLimitDelay(url, duration);
|
||||
});
|
||||
|
||||
return result;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Issue: Requests Too Slow
|
||||
|
||||
**Symptom**: Data collection takes longer than expected
|
||||
|
||||
**Solution**:
|
||||
1. Check rate limit setting: `requests_per_second`
|
||||
2. Increase rate limit if endpoints can handle it
|
||||
3. Monitor endpoint response times
|
||||
4. Consider per-endpoint rate limits
|
||||
|
||||
### Issue: Endpoints Overwhelmed
|
||||
|
||||
**Symptom**: HTTP 429 (Too Many Requests) or timeouts
|
||||
|
||||
**Solution**:
|
||||
1. Decrease `requests_per_second`
|
||||
2. Implement exponential backoff (Phase 1, Task 3.2)
|
||||
3. Add per-endpoint rate limit overrides
|
||||
4. Monitor endpoint health
|
||||
|
||||
### Issue: Uneven Distribution
|
||||
|
||||
**Symptom**: Some endpoints polled more frequently than others
|
||||
|
||||
**Solution**:
|
||||
1. Enable `per_endpoint: true` in configuration
|
||||
2. Set appropriate `rate_limit_override` per endpoint
|
||||
3. Review polling schedule distribution
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
### Planned Enhancements (Post-Phase 1)
|
||||
|
||||
1. **Dynamic Rate Adjustment**: Adjust rate based on endpoint health
|
||||
2. **Adaptive Rate Limiting**: Auto-tune based on response times
|
||||
3. **Token Bucket Size**: Configure burst allowance
|
||||
4. **Rate Limit Warm-up**: Gradual ramp-up after restart
|
||||
5. **Priority-Based Limiting**: Different rates for different data priorities
|
||||
|
||||
### Integration Points
|
||||
|
||||
- **Backpressure Controller** (Phase 1.2): Adjust rate based on buffer usage
|
||||
- **Health Check** (Phase 3.6): Include rate limit statistics
|
||||
- **Configuration Reload** (Future): Hot-reload rate limit changes
|
||||
|
||||
## References
|
||||
|
||||
### Requirements Traceability
|
||||
|
||||
- **Req-FR-16**: Rate limiting for HTTP requests (enhanced)
|
||||
- **Req-Arch-6**: Thread-safe concurrent operations
|
||||
- **Req-NFR-2**: Performance under load
|
||||
|
||||
### Related Documentation
|
||||
|
||||
- [PROJECT_IMPLEMENTATION_PLAN.md](../PROJECT_IMPLEMENTATION_PLAN.md) - Phase 1.1
|
||||
- [system-architecture.md](../architecture/system-architecture.md) - Adapter pattern
|
||||
- [test-strategy.md](../testing/test-strategy.md) - TDD approach
|
||||
|
||||
### External References
|
||||
|
||||
- [Google Guava RateLimiter](https://github.com/google/guava/wiki/RateLimiterExplained)
|
||||
- [Token Bucket Algorithm](https://en.wikipedia.org/wiki/Token_bucket)
|
||||
|
||||
---
|
||||
|
||||
**Document Version**: 1.0
|
||||
**Last Updated**: 2025-11-20
|
||||
**Author**: HSP Development Team
|
||||
**Status**: Implemented (Phase 1.1)
|
||||
Reference in New Issue
Block a user