diff --git a/docs/autoscaling-policies.md b/docs/autoscaling-policies.md new file mode 100644 index 0000000..88b08c9 --- /dev/null +++ b/docs/autoscaling-policies.md @@ -0,0 +1,65 @@ +# Autoscaling Policies + +This document captures the implementation and rollout guidance for performance plan item `Autoscaling policies`. + +Exposed metrics: + +- `/actuator/prometheus` is enabled for scraping. +- `video.clipping.queue.pending` +- `video.clipping.queue.processing` +- `video.clipping.queue.dlq` +- `video.clipping.queue.oldest.pending.age.seconds` + +Worker policy: + +- Start with one FFmpeg process per worker pod. +- Use `minReplicas: 1` and `maxReplicas: 20`. +- Scale out when either queue depth or queue age indicates backlog: +- `video_clipping_queue_pending > 5` per worker target. +- `video_clipping_queue_oldest_pending_age_seconds > 30`. +- Keep CPU as a secondary safeguard with average utilization near `70%`. + +API policy: + +- Use a separate HPA from workers. +- Scale API pods on CPU and request latency or concurrent requests if those metrics are available from the ingress/controller layer. +- Start with `minReplicas: 2`, `maxReplicas: 10`, and CPU target near `60%`. + +Sizing rationale: + +- The worker CPU benchmark and FFmpeg preset benchmark show clip generation is CPU-bound once source staging is local. +- The object-storage bandwidth benchmark shows throughput degradation becomes material below `50 MiB/s`, so queue-age should remain part of the worker policy even when CPU is not yet saturated. +- The visibility-timeout default is conservative, so scaling should respond to queue buildup before stale-claim recovery becomes common. + +Suggested worker HPA or KEDA shape: + +```yaml +apiVersion: keda.sh/v1alpha1 +kind: ScaledObject +metadata: + name: video-clipping-worker +spec: + scaleTargetRef: + name: video-clipping-worker + minReplicaCount: 1 + maxReplicaCount: 20 + cooldownPeriod: 120 + pollingInterval: 15 + triggers: + - type: prometheus + metadata: + serverAddress: http://prometheus.monitoring.svc.cluster.local:9090 + metricName: video_clipping_queue_pending + threshold: "5" + query: sum(video_clipping_queue_pending) + - type: prometheus + metadata: + serverAddress: http://prometheus.monitoring.svc.cluster.local:9090 + metricName: video_clipping_queue_oldest_pending_age_seconds + threshold: "30" + query: max(video_clipping_queue_oldest_pending_age_seconds) +``` + +Operational note: + +- If production uses SQS instead of the database queue, keep the same policy shape but source queue depth and age from native cloud metrics instead of the local Micrometer gauges. diff --git a/docs/video-clipping-service-implementation-plan.md b/docs/video-clipping-service-implementation-plan.md index 1b3ab9a..4150ea3 100644 --- a/docs/video-clipping-service-implementation-plan.md +++ b/docs/video-clipping-service-implementation-plan.md @@ -714,7 +714,7 @@ Keep the domain independent from Spring framework details. Adapters implement st 2. [x] FFmpeg preset benchmarks. 3. [x] Object-storage bandwidth tests. 4. [x] Queue visibility-timeout tuning. -5. [ ] Autoscaling policies. +5. [x] Autoscaling policies. 6. [ ] Cost model by video minute. ### Operational Readiness diff --git a/pom.xml b/pom.xml index 2169766..15a49e8 100644 --- a/pom.xml +++ b/pom.xml @@ -51,6 +51,10 @@ org.springframework.boot spring-boot-starter-actuator + + io.micrometer + micrometer-registry-prometheus + org.flywaydb flyway-core diff --git a/src/main/java/org/example/videoclips/observability/DatabaseQueueMetricsBinder.java b/src/main/java/org/example/videoclips/observability/DatabaseQueueMetricsBinder.java new file mode 100644 index 0000000..6ecbd16 --- /dev/null +++ b/src/main/java/org/example/videoclips/observability/DatabaseQueueMetricsBinder.java @@ -0,0 +1,39 @@ +package org.example.videoclips.observability; + +import io.micrometer.core.instrument.Gauge; +import io.micrometer.core.instrument.MeterRegistry; +import org.example.videoclips.queue.entity.QueueMessageEntity; +import org.example.videoclips.queue.spring.QueueMessageJpaRepository; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.time.Instant; + +@Component +@ConditionalOnProperty(name = "video-clipping.queue", havingValue = "db") +public class DatabaseQueueMetricsBinder { + + public DatabaseQueueMetricsBinder(MeterRegistry meterRegistry, QueueMessageJpaRepository queueMessageJpaRepository) { + Gauge.builder("video.clipping.queue.pending", queueMessageJpaRepository, repository -> repository.countByStatus("PENDING")) + .description("Pending clip-job queue messages") + .register(meterRegistry); + Gauge.builder("video.clipping.queue.processing", queueMessageJpaRepository, repository -> repository.countByStatus("PROCESSING")) + .description("Claimed clip-job queue messages still processing") + .register(meterRegistry); + Gauge.builder("video.clipping.queue.dlq", queueMessageJpaRepository, repository -> repository.countByStatus("DLQ")) + .description("Dead-lettered clip-job queue messages") + .register(meterRegistry); + Gauge.builder("video.clipping.queue.oldest.pending.age.seconds", queueMessageJpaRepository, this::oldestPendingAgeSeconds) + .description("Age in seconds of the oldest pending clip-job queue message") + .register(meterRegistry); + } + + private double oldestPendingAgeSeconds(QueueMessageJpaRepository queueMessageJpaRepository) { + QueueMessageEntity oldestPending = queueMessageJpaRepository.findFirstByStatusOrderByAvailableAtAsc("PENDING"); + if (oldestPending == null) { + return 0.0; + } + return Math.max(0.0, Duration.between(oldestPending.getAvailableAt(), Instant.now()).toSeconds()); + } +} diff --git a/src/main/java/org/example/videoclips/queue/spring/QueueMessageJpaRepository.java b/src/main/java/org/example/videoclips/queue/spring/QueueMessageJpaRepository.java index f53477d..12f8e96 100644 --- a/src/main/java/org/example/videoclips/queue/spring/QueueMessageJpaRepository.java +++ b/src/main/java/org/example/videoclips/queue/spring/QueueMessageJpaRepository.java @@ -10,4 +10,8 @@ public interface QueueMessageJpaRepository extends JpaRepository findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(String status, java.time.Instant availableAt); List findTop10ByStatusAndClaimedAtLessThanEqualOrderByClaimedAtAsc(String status, java.time.Instant claimedAt); + + long countByStatus(String status); + + QueueMessageEntity findFirstByStatusOrderByAvailableAtAsc(String status); } diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 750e1fd..e0187a2 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -1,6 +1,6 @@ spring.application.name=video-clipping-service spring.mvc.problemdetails.enabled=true -management.endpoints.web.exposure.include=health,info,metrics +management.endpoints.web.exposure.include=health,info,metrics,prometheus spring.jpa.open-in-view=false video-clipping.repository=memory video-clipping.storage=memory diff --git a/src/test/java/org/example/videoclips/observability/DatabaseQueueMetricsBinderTest.java b/src/test/java/org/example/videoclips/observability/DatabaseQueueMetricsBinderTest.java new file mode 100644 index 0000000..e52593d --- /dev/null +++ b/src/test/java/org/example/videoclips/observability/DatabaseQueueMetricsBinderTest.java @@ -0,0 +1,36 @@ +package org.example.videoclips.observability; + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.example.videoclips.queue.entity.QueueMessageEntity; +import org.example.videoclips.queue.spring.QueueMessageJpaRepository; +import org.junit.jupiter.api.Test; + +import java.time.Instant; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class DatabaseQueueMetricsBinderTest { + + @Test + void registersQueueDepthAndAgeGauges() { + QueueMessageJpaRepository repository = mock(QueueMessageJpaRepository.class); + when(repository.countByStatus("PENDING")).thenReturn(7L); + when(repository.countByStatus("PROCESSING")).thenReturn(2L); + when(repository.countByStatus("DLQ")).thenReturn(1L); + + QueueMessageEntity oldestPending = new QueueMessageEntity(); + oldestPending.setAvailableAt(Instant.now().minusSeconds(42)); + when(repository.findFirstByStatusOrderByAvailableAtAsc("PENDING")).thenReturn(oldestPending); + + SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); + new DatabaseQueueMetricsBinder(meterRegistry, repository); + + assertEquals(7.0, meterRegistry.get("video.clipping.queue.pending").gauge().value()); + assertEquals(2.0, meterRegistry.get("video.clipping.queue.processing").gauge().value()); + assertEquals(1.0, meterRegistry.get("video.clipping.queue.dlq").gauge().value()); + double oldestPendingAgeSeconds = meterRegistry.get("video.clipping.queue.oldest.pending.age.seconds").gauge().value(); + assertEquals(42.0, oldestPendingAgeSeconds, 2.0); + } +}