mirror of
https://github.com/Dvorinka/Containr.git
synced 2026-10-28 07:41:27 +00:00
GET /api/v1/operations aggregates active deployments, per-service deploy-queue depth via Queue.Snapshot(), 24h deployment failures, cron runs and backups scoped to the caller's projects (admins see all). Frontend /operations page (Operate nav) shows in-flight count, queued depth, failures and backup stats with live refetch while deploys are active. CLI `containr operations` + MCP containr_operations_get. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
177 lines
4.0 KiB
Go
177 lines
4.0 KiB
Go
package deployqueue
|
|
|
|
import (
|
|
"context"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
func TestSerializesPerService(t *testing.T) {
|
|
q := New()
|
|
svc := uuid.New()
|
|
|
|
var active, maxActive int32
|
|
finish := make(chan struct{})
|
|
for i := 0; i < 3; i++ {
|
|
q.Enqueue(svc, Job{Run: func(ctx context.Context) {
|
|
if n := atomic.AddInt32(&active, 1); n > atomic.LoadInt32(&maxActive) {
|
|
atomic.StoreInt32(&maxActive, n)
|
|
}
|
|
<-finish // hold until released
|
|
atomic.AddInt32(&active, -1)
|
|
}})
|
|
}
|
|
|
|
deadline := time.After(2 * time.Second)
|
|
for atomic.LoadInt32(&active) < 1 {
|
|
select {
|
|
case <-deadline:
|
|
t.Fatal("first job never started")
|
|
default:
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
}
|
|
close(finish)
|
|
deadline = time.After(2 * time.Second)
|
|
for atomic.LoadInt32(&active) > 0 {
|
|
select {
|
|
case <-deadline:
|
|
t.Fatal("jobs did not drain")
|
|
default:
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
}
|
|
if got := atomic.LoadInt32(&maxActive); got != 1 {
|
|
t.Fatalf("expected max concurrency 1, got %d", got)
|
|
}
|
|
}
|
|
|
|
func TestDifferentServicesRunInParallel(t *testing.T) {
|
|
q := New()
|
|
started := make(chan struct{}, 2)
|
|
release := make(chan struct{})
|
|
for i := 0; i < 2; i++ {
|
|
q.Enqueue(uuid.New(), Job{Run: func(ctx context.Context) {
|
|
started <- struct{}{}
|
|
<-release
|
|
}})
|
|
}
|
|
for i := 0; i < 2; i++ {
|
|
select {
|
|
case <-started:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("parallel services did not both start")
|
|
}
|
|
}
|
|
close(release)
|
|
}
|
|
|
|
func TestCancelQueuedRemovesJob(t *testing.T) {
|
|
q := New()
|
|
svc := uuid.New()
|
|
release := make(chan struct{})
|
|
q.Enqueue(svc, Job{DeploymentID: uuid.New(), Run: func(ctx context.Context) { <-release }})
|
|
|
|
queuedID := uuid.New()
|
|
ran := make(chan struct{})
|
|
q.Enqueue(svc, Job{DeploymentID: queuedID, Run: func(ctx context.Context) { close(ran) }})
|
|
|
|
if got := q.Cancel(svc, queuedID); got != WasQueued {
|
|
t.Fatalf("expected WasQueued, got %v", got)
|
|
}
|
|
close(release)
|
|
select {
|
|
case <-ran:
|
|
t.Fatal("cancelled queued job still ran")
|
|
case <-time.After(200 * time.Millisecond):
|
|
}
|
|
}
|
|
|
|
func TestCancelActiveCancelsContext(t *testing.T) {
|
|
q := New()
|
|
svc := uuid.New()
|
|
id := uuid.New()
|
|
ctxDone := make(chan struct{})
|
|
q.Enqueue(svc, Job{DeploymentID: id, Run: func(ctx context.Context) {
|
|
<-ctx.Done()
|
|
close(ctxDone)
|
|
}})
|
|
|
|
deadline := time.After(2 * time.Second)
|
|
for {
|
|
q.mu.Lock()
|
|
_, ok := q.cancels[id]
|
|
q.mu.Unlock()
|
|
if ok {
|
|
break
|
|
}
|
|
select {
|
|
case <-deadline:
|
|
t.Fatal("job never registered")
|
|
default:
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
}
|
|
if got := q.Cancel(svc, id); got != WasActive {
|
|
t.Fatalf("expected WasActive, got %v", got)
|
|
}
|
|
select {
|
|
case <-ctxDone:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("active job context not cancelled")
|
|
}
|
|
}
|
|
|
|
func TestSnapshot(t *testing.T) {
|
|
q := New()
|
|
if got := q.Snapshot(); len(got) != 0 {
|
|
t.Fatalf("empty queue snapshot = %v, want empty", got)
|
|
}
|
|
|
|
svc := uuid.New()
|
|
release := make(chan struct{})
|
|
q.Enqueue(svc, Job{Run: func(ctx context.Context) { <-release }})
|
|
q.Enqueue(svc, Job{Run: func(ctx context.Context) {}})
|
|
q.Enqueue(svc, Job{Run: func(ctx context.Context) {}})
|
|
|
|
deadline := time.After(2 * time.Second)
|
|
for {
|
|
snap := q.Snapshot()
|
|
if len(snap) == 1 && snap[0].Running {
|
|
if snap[0].ServiceID != svc {
|
|
t.Fatalf("snapshot service = %v, want %v", snap[0].ServiceID, svc)
|
|
}
|
|
if snap[0].Queued != 2 {
|
|
t.Fatalf("queued = %d, want 2", snap[0].Queued)
|
|
}
|
|
close(release)
|
|
return
|
|
}
|
|
select {
|
|
case <-deadline:
|
|
t.Fatalf("snapshot never showed running job: %v", snap)
|
|
default:
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestEnqueuePosition(t *testing.T) {
|
|
q := New()
|
|
svc := uuid.New()
|
|
release := make(chan struct{})
|
|
if pos := q.Enqueue(svc, Job{Run: func(ctx context.Context) { <-release }}); pos != 0 {
|
|
t.Fatalf("first job position = %d, want 0", pos)
|
|
}
|
|
if pos := q.Enqueue(svc, Job{Run: func(ctx context.Context) {}}); pos != 1 {
|
|
t.Fatalf("second job position = %d, want 1", pos)
|
|
}
|
|
if pos := q.Enqueue(svc, Job{Run: func(ctx context.Context) {}}); pos != 2 {
|
|
t.Fatalf("third job position = %d, want 2", pos)
|
|
}
|
|
close(release)
|
|
}
|