A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
Go
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541package agent
import ( "context" "errors" "fmt" "io" "net/http" "net/http/httptest" "os" "path/filepath" "strings" "sync" "testing" "time"
"connectrpc.com/connect" agentv1 "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/agent/v1" "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/agent/v1/agentv1connect" backupcfg "forgejo.alexma.top/alexma233/composia/internal/core/backup" "forgejo.alexma.top/alexma233/composia/internal/core/config" "forgejo.alexma.top/alexma233/composia/internal/core/task" "forgejo.alexma.top/alexma233/composia/internal/platform/rpcutil" "forgejo.alexma.top/alexma233/composia/internal/platform/store")
func TestResolveComposeForceRecreateAutoDetectsServiceDirBindMount(t *testing.T) { rootDir := t.TempDir() serviceDir := filepath.Join(rootDir, "repo", "demo") binDir := filepath.Join(rootDir, "bin") argsFile := filepath.Join(rootDir, "args.txt") if err := os.MkdirAll(serviceDir, 0o750); err != nil { t.Fatalf("create service dir: %v", err) } if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerPath := filepath.Join(binDir, "docker") script := "#!/bin/sh\nprintf '%s ' \"$@\" > \"$TEST_ARGS_FILE\"\nprintf '%s' \"$TEST_COMPOSE_CONFIG\"\n" if err := os.WriteFile(dockerPath, []byte(script), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_ARGS_FILE", argsFile) t.Setenv("TEST_COMPOSE_CONFIG", fmt.Sprintf(`{"services":{"web":{"volumes":[{"type":"bind","source":%q,"target":"/app/config"},{"type":"volume","source":"named","target":"/data"}]}}}`, filepath.Join(serviceDir, "config")))
var logBuilder strings.Builder forceRecreate, err := resolveComposeForceRecreate(context.Background(), serviceDir, composeCommandConfig{ProjectName: "demo", Files: []string{"compose.yaml", "compose.override.yaml"}}, composeRecreateAuto, func(output string) error { _, writeErr := logBuilder.WriteString(output) return writeErr }) if err != nil { t.Fatalf("resolve compose force recreate: %v", err) } if !forceRecreate { t.Fatal("expected auto mode to force recreate for service-dir bind mount") } logText := logBuilder.String() if !strings.Contains(logText, "detected bind mounts from replaceable service directory") || !strings.Contains(logText, "using docker compose up -d --force-recreate") { t.Fatalf("expected recreate warning log, got %q", logText) } argsContent, err := os.ReadFile(argsFile) //nolint:gosec if err != nil { t.Fatalf("read args file: %v", err) } if got := string(argsContent); got != "compose --project-name demo -f compose.yaml -f compose.override.yaml config --format json " { t.Fatalf("unexpected docker args %q", got) }}
func TestRunComposeUpWithForceRecreatePassesFlag(t *testing.T) { rootDir := t.TempDir() serviceDir := filepath.Join(rootDir, "repo", "demo") binDir := filepath.Join(rootDir, "bin") argsFile := filepath.Join(rootDir, "args.txt") if err := os.MkdirAll(serviceDir, 0o750); err != nil { t.Fatalf("create service dir: %v", err) } if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerPath := filepath.Join(binDir, "docker") script := "#!/bin/sh\nprintf '%s ' \"$@\" > \"$TEST_ARGS_FILE\"\n" if err := os.WriteFile(dockerPath, []byte(script), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_ARGS_FILE", argsFile)
if err := runComposeUpWithOptions(context.Background(), serviceDir, composeCommandConfig{ProjectName: "demo", Files: []string{"compose.yaml", "compose.override.yaml"}}, composeUpOptions{ForceRecreate: true}, func(string) error { return nil }); err != nil { t.Fatalf("run compose up: %v", err) } argsContent, err := os.ReadFile(argsFile) //nolint:gosec if err != nil { t.Fatalf("read args file: %v", err) } if string(argsContent) != "compose --project-name demo -f compose.yaml -f compose.override.yaml up -d --force-recreate " { t.Fatalf("unexpected docker args %q", string(argsContent)) }}
func TestExecuteDeployTaskSkipsComposeForConfigInfraService(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") dockerLogFile := filepath.Join(rootDir, "docker.log") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerPath := filepath.Join(binDir, "docker") script := "#!/bin/sh\nprintf '%s\n' \"$*\" >> \"$TEST_DOCKER_LOG_FILE\"\nexit 97\n" if err := os.WriteFile(dockerPath, []byte(script), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_DOCKER_LOG_FILE", dockerLogFile)
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(cfg.RepoDir, 0o750); err != nil { t.Fatalf("create repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.MkdirAll(cfg.CaddyGeneratedDir(), 0o750); err != nil { t.Fatalf("create caddy generated dir: %v", err) }
bundle := buildBundleArchive(t, map[string]string{ "host-service/composia-meta.yaml": "name: host-service\nnodes:\n - main\ninfra:\n config: {}\nnetwork:\n caddy:\n enabled: true\n source: ./host-service.caddy\n", "host-service/host-service.caddy": "host.example.com { reverse_proxy 127.0.0.1:8080 }\n", }) reportServer := &agentExecutionTestReportServer{} bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{bundle: bundle, expectedTaskID: "task-config", responseServiceName: "host-service", responseRelativeRoot: "host-service"}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { //nolint:goconst return "", errString("unexpected token") } return "main", nil //nolint:goconst }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close()
reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != agentTestMainToken { return "", errString("unexpected token") } return agentTestMainNodeID, nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(agentTestMainToken))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(agentTestMainToken))) logUploader := newTaskLogUploader(reportClient, "task-config") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-config", Type: protoAgentTaskType(task.TypeDeploy), ServiceName: "host-service", NodeId: agentTestMainNodeID, RepoRevision: "deadbeef", ServiceDir: "host-service"} if err := executeDeployTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute infra.config deploy task: %v", err) }
if _, err := os.Stat(dockerLogFile); !errors.Is(err, os.ErrNotExist) { t.Fatalf("expected docker command to be skipped, got stat err=%v", err) } generated, err := os.ReadFile(filepath.Join(cfg.CaddyGeneratedDir(), "host-service.caddy")) if err != nil { t.Fatalf("read generated caddy file: %v", err) } if !strings.Contains(string(generated), "reverse_proxy 127.0.0.1:8080") { t.Fatalf("unexpected generated caddy content %q", string(generated)) }
reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusSucceeded) { t.Fatalf("expected succeeded task status, got %q", reportServer.taskStatus) } if reportServer.runtimeStatus != store.ServiceRuntimeRunning { t.Fatalf("expected running runtime status, got %q", reportServer.runtimeStatus) } if reportServer.stepStatuses[task.StepComposeUp] != string(task.StatusSucceeded) || reportServer.stepStatuses[task.StepCaddySync] != string(task.StatusSucceeded) { t.Fatalf("unexpected step statuses %+v", reportServer.stepStatuses) }}
func TestExecuteStopTaskDownloadsBundleAndRunsComposeDown(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") argsFile := filepath.Join(rootDir, "args.txt") pwdFile := filepath.Join(rootDir, "pwd.txt") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerPath := filepath.Join(binDir, "docker") script := "#!/bin/sh\npwd > \"$TEST_PWD_FILE\"\nprintf '%s ' \"$@\" > \"$TEST_ARGS_FILE\"\n" //nolint:goconst if err := os.WriteFile(dockerPath, []byte(script), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_ARGS_FILE", argsFile) t.Setenv("TEST_PWD_FILE", pwdFile)
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(cfg.RepoDir, 0o750); err != nil { t.Fatalf("create repo dir: %v", err) } if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "backup"), 0o750); err != nil { t.Fatalf("create backup repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "backup", "composia-meta.yaml"), []byte("name: backup\nproject_name: infra-rustic\ncompose_files:\n - compose.yaml\n - compose.backup.yaml\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n profile: prod\n"), 0o600); err != nil { t.Fatalf("write backup meta: %v", err) } if err := os.MkdirAll(cfg.CaddyGeneratedDir(), 0o750); err != nil { t.Fatalf("create caddy generated dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.CaddyGeneratedDir(), "demo.caddy"), []byte("demo"), 0o600); err != nil { t.Fatalf("seed generated caddy file: %v", err) }
bundle := buildBundleArchive(t, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./demo.caddy\n", "demo/docker-compose.yaml": "services: {}\n", "demo/demo.caddy": "demo.example.com { reverse_proxy 127.0.0.1:8080 }\n", }) reportServer := &agentExecutionTestReportServer{} bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{bundle: bundle, expectedTaskID: "task-1"}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close()
reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-1") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{ TaskId: "task-1", Type: protoAgentTaskType(task.TypeStop), ServiceName: "demo", NodeId: "main", RepoRevision: "deadbeef", ServiceDir: "demo", } if err := executeStopTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute stop task: %v", err) }
argsContent, err := os.ReadFile(argsFile) //nolint:gosec if err != nil { t.Fatalf("read args file: %v", err) } if string(argsContent) != "compose --project-name demo down " { t.Fatalf("unexpected docker args %q", string(argsContent)) } pwdContent, err := os.ReadFile(pwdFile) //nolint:gosec if err != nil { t.Fatalf("read pwd file: %v", err) } if string(bytesTrimSpace(pwdContent)) != filepath.Join(cfg.RepoDir, "demo") { t.Fatalf("expected docker cwd %q, got %q", filepath.Join(cfg.RepoDir, "demo"), string(bytesTrimSpace(pwdContent))) } if _, err := os.Stat(filepath.Join(cfg.CaddyGeneratedDir(), "demo.caddy")); !errors.Is(err, os.ErrNotExist) { t.Fatalf("expected generated caddy file removed, got stat err=%v", err) }
reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusSucceeded) { t.Fatalf("expected succeeded task status, got %q", reportServer.taskStatus) } if reportServer.runtimeStatus != store.ServiceRuntimeStopped { t.Fatalf("expected stopped runtime status, got %q", reportServer.runtimeStatus) } if len(reportServer.stepStatuses) == 0 || reportServer.stepStatuses[task.StepRender] != string(task.StatusSucceeded) || reportServer.stepStatuses[task.StepComposeDown] != string(task.StatusSucceeded) || reportServer.stepStatuses[task.StepCaddySync] != string(task.StatusSucceeded) { t.Fatalf("unexpected step statuses %+v", reportServer.stepStatuses) }}
func TestExecuteBackupTaskRunsRusticAndReportsSnapshot(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") dockerArgsFile := filepath.Join(rootDir, "docker-args.txt") dockerPath := filepath.Join(binDir, "docker") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerScript := "#!/bin/sh\nprintf '%s\n' \"$*\" >> \"$TEST_DOCKER_ARGS_FILE\"\nprintf '[INFO] snapshot abc12345 successfully saved.\\n'\n" if err := os.WriteFile(dockerPath, []byte(dockerScript), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_DOCKER_ARGS_FILE", dockerArgsFile)
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(cfg.RepoDir, 0o750); err != nil { t.Fatalf("create repo dir: %v", err) } if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "backup"), 0o750); err != nil { t.Fatalf("create backup repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "backup", "composia-meta.yaml"), []byte("name: backup\nproject_name: infra-rustic\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n"), 0o600); err != nil { t.Fatalf("write backup meta: %v", err) } serviceBundle := buildBundleArchive(t, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\n", "demo/config/app.env": "HELLO=world\n", "demo/.composia-backup.json": `{"rustic":{"service_name":"backup","service_dir":"backup","compose_service":"rustic","profile":"prod","data_protect_dir":"/data-protect","node_id":"main"},"items":[{"name":"config","strategy":"files.copy","include":["./config"],"provider":"rustic","tags":["composia-service:demo","composia-data:config","composia-node:main"]}]}`, }) rusticBundle := buildBundleArchive(t, map[string]string{ "backup/composia-meta.yaml": "name: backup\nproject_name: infra-rustic\ncompose_files:\n - compose.yaml\n - compose.backup.yaml\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n profile: prod\n data_protect_dir: /data-protect\n", }) reportServer := &agentExecutionTestReportServer{} bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{expectedTaskID: "task-backup", bundlesByServiceDir: map[string]bundleTestResponse{"": {bundle: serviceBundle, serviceName: "demo", relativeRoot: "demo"}, "backup": {bundle: rusticBundle, serviceName: "backup", relativeRoot: "backup"}}}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close()
reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-backup") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-backup", Type: protoAgentTaskType(task.TypeBackup), ServiceName: "demo", NodeId: "main", RepoRevision: "deadbeef", ServiceDir: "demo", DataNames: []string{"config"}} if err := executeBackupTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute backup task: %v", err) }
argsContent, err := os.ReadFile(dockerArgsFile) //nolint:gosec if err != nil { t.Fatalf("read docker args file: %v", err) } argsLog := string(argsContent) if !strings.Contains(argsLog, "compose --project-name infra-rustic -f compose.yaml -f compose.backup.yaml run --rm -v") || !strings.Contains(argsLog, "rustic -P prod backup --host main") { t.Fatalf("unexpected rustic args %q", string(argsContent)) } if !strings.Contains(argsLog, "/data-protect/") { t.Fatalf("expected rustic to receive mapped data-protect path, got %q", argsLog) } if strings.Contains(argsLog, cfg.StateDir+"/") { t.Fatalf("expected rustic args to avoid raw state_dir path, got %q", argsLog) } if !strings.Contains(argsLog, "--tag composia-service:demo --tag composia-data:config --tag composia-node:main") { t.Fatalf("expected backup tags in docker args, got %q", argsLog) } reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusSucceeded) { t.Fatalf("expected succeeded task status, got %q", reportServer.taskStatus) } if reportServer.backupArtifactRef != "abc12345" { t.Fatalf("expected backup artifact ref abc12345, got %q", reportServer.backupArtifactRef) } if reportServer.backupDataName != "config" { t.Fatalf("expected backup data name config, got %q", reportServer.backupDataName) } if reportServer.backupStatus != string(task.StatusSucceeded) { t.Fatalf("expected succeeded backup status, got %q", reportServer.backupStatus) }}
func TestExecuteCaddySyncTaskCopiesServiceCaddyFile(t *testing.T) { rootDir := t.TempDir() cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(cfg.RepoDir, 0o750); err != nil { t.Fatalf("create repo dir: %v", err) } if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "backup"), 0o750); err != nil { t.Fatalf("create backup repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "backup", "composia-meta.yaml"), []byte("name: backup\nproject_name: infra-rustic\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n"), 0o600); err != nil { t.Fatalf("write backup meta: %v", err) }
bundle := buildBundleArchive(t, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./demo.caddy\n", "demo/docker-compose.yaml": "services: {}\n", "demo/demo.caddy": "demo.example.com { reverse_proxy 127.0.0.1:8080 }\n", }) reportServer := &agentExecutionTestReportServer{} bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{bundle: bundle, expectedTaskID: "task-caddy-sync"}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close()
reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-caddy-sync") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-caddy-sync", Type: protoAgentTaskType(task.TypeCaddySync), ServiceName: "demo", NodeId: "main", RepoRevision: "deadbeef", ServiceDir: "demo"} if err := executeCaddySyncTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute caddy sync task: %v", err) } content, err := os.ReadFile(filepath.Join(cfg.CaddyGeneratedDir(), "demo.caddy")) if err != nil { t.Fatalf("read generated caddy file: %v", err) } if string(content) != "demo.example.com { reverse_proxy 127.0.0.1:8080 }\n" { t.Fatalf("unexpected generated caddy file %q", string(content)) } reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.stepStatuses[task.StepCaddySync] != string(task.StatusSucceeded) { t.Fatalf("expected caddy_sync step succeeded, got %+v", reportServer.stepStatuses) }}
func TestExecuteCaddyTasksUseServiceDirForGeneratedFileName(t *testing.T) { rootDir := t.TempDir() cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(cfg.RepoDir, 0o750); err != nil { t.Fatalf("create repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "element-web"), 0o750); err != nil { t.Fatalf("create service dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "element-web", "composia-meta.yaml"), []byte("name: element\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./site-config.caddy\n"), 0o600); err != nil { t.Fatalf("write service meta: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "element-web", "site-config.caddy"), []byte("element.alexma.top { reverse_proxy 127.0.0.1:8080 }\n"), 0o600); err != nil { t.Fatalf("write caddy source: %v", err) } if err := os.MkdirAll(cfg.CaddyGeneratedDir(), 0o750); err != nil { t.Fatalf("create caddy generated dir: %v", err) } if err := syncServiceCaddyFile(context.Background(), cfg, "element-web", filepath.Join(cfg.RepoDir, "element-web"), func(string) error { return nil }); err != nil { t.Fatalf("sync generated caddy file by service dir: %v", err) } if _, err := os.Stat(filepath.Join(cfg.CaddyGeneratedDir(), "element-web.caddy")); err != nil { t.Fatalf("expected element-web.caddy: %v", err) } if _, err := os.Stat(filepath.Join(cfg.CaddyGeneratedDir(), "element.caddy")); !errors.Is(err, os.ErrNotExist) { t.Fatalf("expected element.caddy absent, got stat err=%v", err) } if err := removeServiceCaddyFile(context.Background(), cfg, "element-web", func(string) error { return nil }); err != nil { t.Fatalf("remove generated caddy file by service dir: %v", err) } if _, err := os.Stat(filepath.Join(cfg.CaddyGeneratedDir(), "element-web.caddy")); !errors.Is(err, os.ErrNotExist) { t.Fatalf("expected element-web.caddy removed, got stat err=%v", err) }}
func TestBuildBackupVolumeFlagsServicePath(t *testing.T) { t.Parallel()
serviceRoot := t.TempDir() sourceDir := filepath.Join(serviceRoot, "volumes", "data") if err := os.MkdirAll(sourceDir, 0o750); err != nil { t.Fatalf("create source dir: %v", err) } if err := os.WriteFile(filepath.Join(sourceDir, "hello.txt"), []byte("hello\n"), 0o600); err != nil { t.Fatalf("write source file: %v", err) }
item := backupcfg.RuntimeItem{Name: "db", Strategy: "files.copy", Include: []string{"./volumes/data"}} flags, err := buildBackupVolumeFlags(serviceRoot, "/data-protect/backup-xxx", item) if err != nil { t.Fatalf("build backup volume flags: %v", err) } if len(flags) != 2 || flags[0] != "-v" || !strings.HasSuffix(flags[1], ":ro") { t.Fatalf("unexpected flags: %v", flags) } if !strings.Contains(flags[1], sourceDir+":/data-protect/backup-xxx/paths/volumes_data:ro") { t.Fatalf("expected service path bind mount, got %q", flags[1]) }}
func TestBuildBackupVolumeFlagsDockerVolume(t *testing.T) { t.Parallel()
item := backupcfg.RuntimeItem{Name: "db", Strategy: "files.copy", Include: []string{"caddy_caddy_data"}} flags, err := buildBackupVolumeFlags(t.TempDir(), "/data-protect/backup-xxx", item) if err != nil { t.Fatalf("build backup volume flags: %v", err) } if len(flags) != 2 || flags[0] != "-v" { t.Fatalf("unexpected flags: %v", flags) } expected := "caddy_caddy_data:/data-protect/backup-xxx/volumes/caddy_caddy_data:ro" if flags[1] != expected { t.Fatalf("expected %q, got %q", expected, flags[1]) }}
func TestBuildBackupVolumeFlagsRejectsPathOutsideServiceRoot(t *testing.T) { t.Parallel()
item := backupcfg.RuntimeItem{Name: "db", Strategy: "files.copy", Include: []string{"../secrets"}} _, err := buildBackupVolumeFlags(t.TempDir(), "/data-protect/backup-xxx", item) if err == nil || !strings.Contains(err.Error(), "must stay within the service root") { t.Fatalf("expected service root validation error, got %v", err) }}
func TestBuildBackupVolumeFlagsRejectsMissingServicePath(t *testing.T) { t.Parallel()
serviceRoot := t.TempDir() item := backupcfg.RuntimeItem{Name: "db", Strategy: "files.copy", Include: []string{"./missing"}} _, err := buildBackupVolumeFlags(serviceRoot, "/data-protect/backup-xxx", item) if err == nil || !strings.Contains(err.Error(), "does not exist") { t.Fatalf("expected missing service path error, got %v", err) }}
func TestPrepareRestoreVolumeFlagsServicePath(t *testing.T) { t.Parallel()
serviceRoot := t.TempDir() stagingDir := t.TempDir() targetDir := filepath.Join(serviceRoot, "volumes", "data") if err := os.MkdirAll(filepath.Join(targetDir, "nested"), 0o750); err != nil { t.Fatalf("create target dir: %v", err) } if err := os.WriteFile(filepath.Join(targetDir, "nested", "old.txt"), []byte("old\n"), 0o600); err != nil { t.Fatalf("write old file: %v", err) }
item := backupcfg.RestoreItem{Name: "db", Strategy: "files.copy", Include: []string{"./volumes/data"}, ArtifactRef: "aaa999"} flags, err := prepareRestoreVolumeFlags(context.Background(), serviceRoot, stagingDir, "/data-protect/restore-xxx", item) if err != nil { t.Fatalf("prepare restore volume flags: %v", err) } if len(flags) != 2 || flags[0] != "-v" { t.Fatalf("unexpected flags: %v", flags) } expected := targetDir + ":/data-protect/restore-xxx/db/paths/volumes_data" if flags[1] != expected { t.Fatalf("expected %q, got %q", expected, flags[1]) } if _, err := os.Stat(filepath.Join(targetDir, "nested", "old.txt")); !errors.Is(err, os.ErrNotExist) { t.Fatalf("expected old file removed, got %v", err) } if _, err := os.Stat(filepath.Join(stagingDir, "db", "paths")); err != nil { t.Fatalf("expected restore staging path dirs: %v", err) }}
func TestApplyRestoreItemRejectsFilesCopy(t *testing.T) { t.Parallel()
item := backupcfg.RestoreItem{Name: "db", Strategy: "files.copy"} err := applyRestoreItem(context.Background(), t.TempDir(), t.TempDir(), item, nil) if err == nil { t.Fatal("expected error for files.copy strategy") }}
func TestPrepareRestoreVolumeFlagsServiceFile(t *testing.T) { t.Parallel()
serviceRoot := t.TempDir() stagingDir := t.TempDir() targetPath := filepath.Join(serviceRoot, "config", "app.env") if err := os.MkdirAll(filepath.Dir(targetPath), 0o750); err != nil { t.Fatalf("create target parent: %v", err) } if err := os.WriteFile(targetPath, []byte("old\n"), 0o600); err != nil { t.Fatalf("write old file: %v", err) }
item := backupcfg.RestoreItem{Name: "config", Strategy: "files.copy", Include: []string{"./config/app.env"}, ArtifactRef: "aaa999"} flags, err := prepareRestoreVolumeFlags(context.Background(), serviceRoot, stagingDir, "/data-protect/restore-xxx", item) if err != nil { t.Fatalf("prepare restore volume flags: %v", err) } expected := targetPath + ":/data-protect/restore-xxx/config/paths/config_app.env" if len(flags) != 2 || flags[1] != expected { t.Fatalf("expected %q, got %v", expected, flags) } info, err := os.Stat(targetPath) if err != nil { t.Fatalf("expected target file placeholder: %v", err) } if info.IsDir() { t.Fatal("expected target file placeholder, got directory") }}
func TestRestoreRuntimeItemDirectMountsItemPath(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") dockerLogFile := filepath.Join(rootDir, "docker.log") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerPath := filepath.Join(binDir, "docker") script := "#!/bin/sh\nprintf '%s\n' \"$*\" >> \"$TEST_DOCKER_LOG_FILE\"\nexit 0\n" if err := os.WriteFile(dockerPath, []byte(script), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_DOCKER_LOG_FILE", dockerLogFile)
cfg := &config.AgentConfig{StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } serviceRoot := filepath.Join(rootDir, "repo", "demo") targetDir := filepath.Join(serviceRoot, "config") if err := os.MkdirAll(targetDir, 0o750); err != nil { t.Fatalf("create restore target: %v", err) } if err := os.WriteFile(filepath.Join(targetDir, "stale.txt"), []byte("stale\n"), 0o600); err != nil { t.Fatalf("write stale file: %v", err) } rusticRoot := filepath.Join(rootDir, "repo", "backup") if err := os.MkdirAll(rusticRoot, 0o750); err != nil { t.Fatalf("create rustic root: %v", err) } if err := os.WriteFile(filepath.Join(rusticRoot, "composia-meta.yaml"), []byte("name: backup\nproject_name: infra-rustic\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n"), 0o600); err != nil { t.Fatalf("write rustic meta: %v", err) }
item := backupcfg.RestoreItem{Name: "config", Strategy: "files.copy", Include: []string{"./config"}, ArtifactRef: "abc123"} rustic := &backupcfg.RusticConfig{ServiceName: "backup", ComposeService: "rustic", DataProtectDir: "/data-protect"} if err := restoreRuntimeItem(context.Background(), cfg, serviceRoot, rusticRoot, "task-restore", item, rustic, nil); err != nil { t.Fatalf("restore runtime item: %v", err) }
if _, err := os.Stat(filepath.Join(targetDir, "stale.txt")); !errors.Is(err, os.ErrNotExist) { t.Fatalf("expected stale target cleared, got %v", err) } dockerLog, err := os.ReadFile(dockerLogFile) //nolint:gosec if err != nil { t.Fatalf("read docker log: %v", err) } logText := string(dockerLog) if !strings.Contains(logText, "run --rm -v ") || !strings.Contains(logText, targetDir+":/data-protect/restore-") || !strings.Contains(logText, "/config/paths/config") || !strings.Contains(logText, " rustic restore abc123 /data-protect/restore-") { t.Fatalf("expected direct restore bind mount under item path, got %q", logText) }}
func TestExecuteBackupTaskStopsComposeForTarAfterStop(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") dockerLogFile := filepath.Join(rootDir, "docker.log") dockerPath := filepath.Join(binDir, "docker") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerScript := "#!/bin/sh\nprintf '%s\n' \"$*\" >> \"$TEST_DOCKER_LOG_FILE\"\ncase \"$*\" in *\" rustic backup \"*) printf 'snapshot aaa999 saved\\n' ;; esac\n" if err := os.WriteFile(dockerPath, []byte(dockerScript), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_DOCKER_LOG_FILE", dockerLogFile)
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(cfg.RepoDir, 0o750); err != nil { t.Fatalf("create repo dir: %v", err) } if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "backup"), 0o750); err != nil { t.Fatalf("create backup repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "backup", "composia-meta.yaml"), []byte("name: backup\nproject_name: infra-rustic\ncompose_files:\n - compose.backup.yaml\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n"), 0o600); err != nil { t.Fatalf("write backup meta: %v", err) } serviceBundle := buildBundleArchive(t, map[string]string{ "demo/composia-meta.yaml": "name: demo\ncompose_files:\n - compose.yaml\nnodes:\n - main\n", "demo/config/app.env": "HELLO=world\n", "demo/.composia-backup.json": `{"rustic":{"service_name":"backup","service_dir":"backup","compose_service":"rustic","data_protect_dir":"/data-protect","node_id":"main"},"items":[{"name":"config","strategy":"files.copy_after_stop","include":["./config"],"provider":"rustic"}]}`, }) rusticBundle := buildBundleArchive(t, map[string]string{ "backup/composia-meta.yaml": "name: backup\nproject_name: infra-rustic\ncompose_files:\n - compose.backup.yaml\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n data_protect_dir: /data-protect\n", }) reportServer := &agentExecutionTestReportServer{} bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{expectedTaskID: "task-tar", bundlesByServiceDir: map[string]bundleTestResponse{"": {bundle: serviceBundle, serviceName: "demo", relativeRoot: "demo"}, "backup": {bundle: rusticBundle, serviceName: "backup", relativeRoot: "backup"}}}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close() reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close() bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-tar") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-tar", Type: protoAgentTaskType(task.TypeBackup), ServiceName: "demo", NodeId: "main", RepoRevision: "deadbeef", ServiceDir: "demo", DataNames: []string{"config"}} if err := executeBackupTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute tar_after_stop backup task: %v", err) } dockerLog, err := os.ReadFile(dockerLogFile) //nolint:gosec if err != nil { t.Fatalf("read docker log: %v", err) } if !strings.Contains(string(dockerLog), "compose --project-name demo -f compose.yaml down") || !strings.Contains(string(dockerLog), "compose --project-name demo -f compose.yaml up -d") || !strings.Contains(string(dockerLog), "compose --project-name infra-rustic -f compose.backup.yaml run --rm -v") || !strings.Contains(string(dockerLog), "rustic backup --host main") { t.Fatalf("expected compose down and up around backup, got %q", string(dockerLog)) } if !strings.Contains(string(dockerLog), "/data-protect/") { t.Fatalf("expected mapped data-protect path in rustic backup command, got %q", string(dockerLog)) } reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.backupArtifactRef != "aaa999" { t.Fatalf("expected tar backup artifact aaa999, got %q", reportServer.backupArtifactRef) }}
func TestExecuteBackupTaskRunsPGDumpAll(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") dockerLogFile := filepath.Join(rootDir, "docker.log") dockerPath := filepath.Join(binDir, "docker") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerScript := "#!/bin/sh\nprintf '%s\n' \"$*\" >> \"$TEST_DOCKER_LOG_FILE\"\ncase \"$*\" in *\"postgres pg_dumpall\"*) printf 'dump-sql\\n' ;; *\" rustic backup \"*) printf 'snapshot abc888 saved\\n' ;; esac\n" if err := os.WriteFile(dockerPath, []byte(dockerScript), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_DOCKER_LOG_FILE", dockerLogFile)
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(cfg.RepoDir, 0o750); err != nil { t.Fatalf("create repo dir: %v", err) } if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "backup"), 0o750); err != nil { t.Fatalf("create backup repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "backup", "composia-meta.yaml"), []byte("name: backup\nproject_name: infra-rustic\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n"), 0o600); err != nil { t.Fatalf("write backup meta: %v", err) } serviceBundle := buildBundleArchive(t, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\n", "demo/.composia-backup.json": `{"rustic":{"service_name":"backup","service_dir":"backup","compose_service":"rustic","data_protect_dir":"/data-protect","node_id":"main"},"items":[{"name":"db","strategy":"database.pgdumpall","service":"postgres","provider":"rustic"}]}`, }) rusticBundle := buildBundleArchive(t, map[string]string{ "backup/composia-meta.yaml": "name: backup\nproject_name: infra-rustic\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n data_protect_dir: /data-protect\n", }) reportServer := &agentExecutionTestReportServer{} bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{expectedTaskID: "task-pg", bundlesByServiceDir: map[string]bundleTestResponse{"": {bundle: serviceBundle, serviceName: "demo", relativeRoot: "demo"}, "backup": {bundle: rusticBundle, serviceName: "backup", relativeRoot: "backup"}}}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close() reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close() bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-pg") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-pg", Type: protoAgentTaskType(task.TypeBackup), ServiceName: "demo", NodeId: "main", RepoRevision: "deadbeef", ServiceDir: "demo", DataNames: []string{"db"}} if err := executeBackupTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute pgdumpall backup task: %v", err) } dockerLog, err := os.ReadFile(dockerLogFile) //nolint:gosec if err != nil { t.Fatalf("read docker log: %v", err) } if !strings.Contains(string(dockerLog), "compose --project-name demo exec -T postgres pg_dumpall") { t.Fatalf("expected pg_dumpall compose exec, got %q", string(dockerLog)) } if !strings.Contains(string(dockerLog), "/data-protect/") { t.Fatalf("expected mapped data-protect path in rustic backup command, got %q", string(dockerLog)) } reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.backupArtifactRef != "abc888" { t.Fatalf("expected pg backup artifact abc888, got %q", reportServer.backupArtifactRef) }}
func TestExecuteBackupTaskReportsFailedBackupItem(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") dockerPath := filepath.Join(binDir, "docker") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerScript := "#!/bin/sh\nprintf 'backup failed\\n' >&2\nexit 44\n" if err := os.WriteFile(dockerPath, []byte(dockerScript), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH"))
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(cfg.RepoDir, 0o750); err != nil { t.Fatalf("create repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } serviceBundle := buildBundleArchive(t, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\n", "demo/config/app.env": "HELLO=world\n", "demo/.composia-backup.json": `{"rustic":{"service_name":"backup","service_dir":"backup","compose_service":"rustic","data_protect_dir":"/data-protect","node_id":"main"},"items":[{"name":"config","strategy":"files.copy","include":["./config"],"provider":"rustic"}]}`, }) rusticBundle := buildBundleArchive(t, map[string]string{ "backup/composia-meta.yaml": "name: backup\nproject_name: infra-rustic\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n data_protect_dir: /data-protect\n", }) reportServer := &agentExecutionTestReportServer{} bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{expectedTaskID: "task-backup-fail", bundlesByServiceDir: map[string]bundleTestResponse{"": {bundle: serviceBundle, serviceName: "demo", relativeRoot: "demo"}, "backup": {bundle: rusticBundle, serviceName: "backup", relativeRoot: "backup"}}}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close() reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close() bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-backup-fail") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-backup-fail", Type: protoAgentTaskType(task.TypeBackup), ServiceName: "demo", NodeId: "main", RepoRevision: "deadbeef", ServiceDir: "demo", DataNames: []string{"config"}} if err := executeBackupTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err == nil { t.Fatalf("expected backup task to fail") } reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusFailed) { t.Fatalf("expected failed task status, got %q", reportServer.taskStatus) } if reportServer.backupStatus != string(task.StatusFailed) { t.Fatalf("expected failed backup status, got %q", reportServer.backupStatus) } if reportServer.backupErrorSummary == "" { t.Fatalf("expected backup error summary to be recorded") }}
func TestRusticDataProtectPathMapsConfiguredContainerDir(t *testing.T) { t.Parallel()
cfg := &config.AgentConfig{StateDir: "/srv/composia/state-agent"} rustic := &backupcfg.RusticConfig{DataProtectDir: "/data-protect"} localPath := "/srv/composia/state-agent/data-protect/backup-task-config-123"
mappedPath, err := rusticDataProtectPath(localPath, cfg, rustic) if err != nil { t.Fatalf("map rustic data-protect path: %v", err) } if mappedPath != "/data-protect/backup-task-config-123" { t.Fatalf("expected mapped path %q, got %q", "/data-protect/backup-task-config-123", mappedPath) }}
func TestRusticDataProtectPathRejectsPathsOutsideStageRoot(t *testing.T) { t.Parallel()
cfg := &config.AgentConfig{StateDir: "/srv/composia/state-agent"} rustic := &backupcfg.RusticConfig{DataProtectDir: "/data-protect"}
_, err := rusticDataProtectPath("/srv/composia/state-agent/other/file", cfg, rustic) if err == nil || !strings.Contains(err.Error(), "outside agent data-protect stage root") { t.Fatalf("expected outside stage root error, got %v", err) }}
func TestExecuteCaddyReloadTaskRunsComposeExec(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") argsFile := filepath.Join(rootDir, "args.txt") pwdFile := filepath.Join(rootDir, "pwd.txt") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerPath := filepath.Join(binDir, "docker") script := "#!/bin/sh\npwd > \"$TEST_PWD_FILE\"\nprintf '%s ' \"$@\" > \"$TEST_ARGS_FILE\"\n" //nolint:goconst if err := os.WriteFile(dockerPath, []byte(script), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_ARGS_FILE", argsFile) t.Setenv("TEST_PWD_FILE", pwdFile)
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "caddy"), 0o750); err != nil { t.Fatalf("create repo caddy dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "caddy", "composia-meta.yaml"), []byte("name: edge-proxy\nproject_name: infra-caddy\ncompose_files:\n - compose.yaml\n - compose.edge.yaml\nnodes:\n - main\ninfra:\n caddy:\n compose_service: edge\n config_dir: /etc/caddy\n"), 0o600); err != nil { t.Fatalf("write caddy meta: %v", err) }
reportServer := &agentExecutionTestReportServer{} reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-caddy-reload") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-caddy-reload", Type: protoAgentTaskType(task.TypeCaddyReload), ServiceName: "edge-proxy", NodeId: "main", ServiceDir: "caddy"} if err := executeCaddyReloadTask(context.Background(), reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute caddy reload task: %v", err) }
argsContent, err := os.ReadFile(argsFile) //nolint:gosec if err != nil { t.Fatalf("read args file: %v", err) } if string(argsContent) != "compose --project-name infra-caddy -f compose.yaml -f compose.edge.yaml exec -T edge caddy reload --config /etc/caddy/Caddyfile --adapter caddyfile " { t.Fatalf("unexpected docker args %q", string(argsContent)) } pwdContent, err := os.ReadFile(pwdFile) //nolint:gosec if err != nil { t.Fatalf("read pwd file: %v", err) } if string(bytesTrimSpace(pwdContent)) != filepath.Join(cfg.RepoDir, "caddy") { t.Fatalf("expected docker cwd %q, got %q", filepath.Join(cfg.RepoDir, "caddy"), string(bytesTrimSpace(pwdContent))) }
reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusSucceeded) { t.Fatalf("expected succeeded task status, got %q", reportServer.taskStatus) } if reportServer.stepStatuses[task.StepCaddyReload] != string(task.StatusSucceeded) { t.Fatalf("expected caddy_reload step succeeded, got %+v", reportServer.stepStatuses) }}
func TestExecuteRusticForgetTaskRunsComposeRun(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") argsFile := filepath.Join(rootDir, "args.txt") pwdFile := filepath.Join(rootDir, "pwd.txt") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerPath := filepath.Join(binDir, "docker") script := "#!/bin/sh\npwd > \"$TEST_PWD_FILE\"\nprintf '%s ' \"$@\" > \"$TEST_ARGS_FILE\"\n" //nolint:goconst if err := os.WriteFile(dockerPath, []byte(script), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_ARGS_FILE", argsFile) t.Setenv("TEST_PWD_FILE", pwdFile)
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "backup"), 0o750); err != nil { t.Fatalf("create backup repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "backup", "composia-meta.yaml"), []byte("name: backup\nproject_name: infra-rustic\ncompose_files:\n - compose.yaml\n - compose.ops.yaml\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n profile: prod\n"), 0o600); err != nil { t.Fatalf("write backup meta: %v", err) } bundle := buildBundleArchive(t, map[string]string{ "backup/composia-meta.yaml": "name: backup\nproject_name: infra-rustic\ncompose_files:\n - compose.yaml\n - compose.ops.yaml\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n profile: prod\n", }) bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{bundle: bundle, expectedTaskID: "task-rustic-forget", responseServiceName: "backup", responseRelativeRoot: "backup"}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close()
reportServer := &agentExecutionTestReportServer{} reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-rustic-forget") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-rustic-forget", Type: protoAgentTaskType(task.TypeRusticForget), ServiceName: "backup", NodeId: "main", ParamsJson: `{"service_name":"demo","data_name":"db","service_dir":"backup"}`} if err := executeRusticForgetTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute rustic forget task: %v", err) }
argsContent, err := os.ReadFile(argsFile) //nolint:gosec if err != nil { t.Fatalf("read args file: %v", err) } if got := string(argsContent); got != "compose --project-name infra-rustic -f compose.yaml -f compose.ops.yaml run --rm rustic -P prod forget --filter-host main --filter-tags composia-service:demo --filter-tags composia-data:db " { t.Fatalf("unexpected docker args %q", got) } pwdContent, err := os.ReadFile(pwdFile) //nolint:gosec if err != nil { t.Fatalf("read pwd file: %v", err) } if string(bytesTrimSpace(pwdContent)) != filepath.Join(cfg.RepoDir, "backup") { t.Fatalf("expected docker cwd %q, got %q", filepath.Join(cfg.RepoDir, "backup"), string(bytesTrimSpace(pwdContent))) } reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusSucceeded) { t.Fatalf("expected succeeded task status, got %q", reportServer.taskStatus) } if reportServer.stepStatuses[task.StepPrune] != string(task.StatusSucceeded) { t.Fatalf("expected prune step succeeded, got %+v", reportServer.stepStatuses) }}
func TestExecuteRusticPruneTaskRunsComposeRun(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") argsFile := filepath.Join(rootDir, "args.txt") pwdFile := filepath.Join(rootDir, "pwd.txt") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerPath := filepath.Join(binDir, "docker") script := "#!/bin/sh\npwd > \"$TEST_PWD_FILE\"\nprintf '%s ' \"$@\" > \"$TEST_ARGS_FILE\"\n" //nolint:goconst if err := os.WriteFile(dockerPath, []byte(script), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_ARGS_FILE", argsFile) t.Setenv("TEST_PWD_FILE", pwdFile)
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "backup"), 0o750); err != nil { t.Fatalf("create backup repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } if err := os.WriteFile(filepath.Join(cfg.RepoDir, "backup", "composia-meta.yaml"), []byte("name: backup\nproject_name: infra-rustic\ncompose_files:\n - compose.yaml\n - compose.ops.yaml\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n profile: prod\n"), 0o600); err != nil { t.Fatalf("write backup meta: %v", err) } bundle := buildBundleArchive(t, map[string]string{ "backup/composia-meta.yaml": "name: backup\nproject_name: infra-rustic\ncompose_files:\n - compose.yaml\n - compose.ops.yaml\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n profile: prod\n", }) bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{bundle: bundle, expectedTaskID: "task-rustic-prune", responseServiceName: "backup", responseRelativeRoot: "backup"}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close()
reportServer := &agentExecutionTestReportServer{} reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-rustic-prune") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-rustic-prune", Type: protoAgentTaskType(task.TypeRusticPrune), ServiceName: "backup", NodeId: "main", ParamsJson: `{"service_dir":"backup"}`} if err := executeRusticPruneTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute rustic prune task: %v", err) }
argsContent, err := os.ReadFile(argsFile) //nolint:gosec if err != nil { t.Fatalf("read args file: %v", err) } if got := string(argsContent); got != "compose --project-name infra-rustic -f compose.yaml -f compose.ops.yaml run --rm rustic -P prod prune " { t.Fatalf("unexpected docker args %q", got) } pwdContent, err := os.ReadFile(pwdFile) //nolint:gosec if err != nil { t.Fatalf("read pwd file: %v", err) } if string(bytesTrimSpace(pwdContent)) != filepath.Join(cfg.RepoDir, "backup") { t.Fatalf("expected docker cwd %q, got %q", filepath.Join(cfg.RepoDir, "backup"), string(bytesTrimSpace(pwdContent))) } reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusSucceeded) { t.Fatalf("expected succeeded task status, got %q", reportServer.taskStatus) } if reportServer.stepStatuses[task.StepPrune] != string(task.StatusSucceeded) { t.Fatalf("expected prune step succeeded, got %+v", reportServer.stepStatuses) }}
func TestExecuteRusticInitTaskRunsComposeRun(t *testing.T) { rootDir := t.TempDir() binDir := filepath.Join(rootDir, "bin") argsFile := filepath.Join(rootDir, "args.txt") pwdFile := filepath.Join(rootDir, "pwd.txt") if err := os.MkdirAll(binDir, 0o750); err != nil { t.Fatalf("create bin dir: %v", err) } dockerPath := filepath.Join(binDir, "docker") script := "#!/bin/sh\npwd > \"$TEST_PWD_FILE\"\nprintf '%s ' \"$@\" > \"$TEST_ARGS_FILE\"\n" //nolint:goconst if err := os.WriteFile(dockerPath, []byte(script), 0o750); err != nil { //nolint:gosec t.Fatalf("write fake docker script: %v", err) } t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("TEST_ARGS_FILE", argsFile) t.Setenv("TEST_PWD_FILE", pwdFile)
cfg := &config.AgentConfig{RepoDir: filepath.Join(rootDir, "repo"), StateDir: filepath.Join(rootDir, "state")} if err := os.MkdirAll(filepath.Join(cfg.RepoDir, "backup"), 0o750); err != nil { t.Fatalf("create backup repo dir: %v", err) } if err := os.MkdirAll(cfg.StateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) } bundle := buildBundleArchive(t, map[string]string{ "backup/composia-meta.yaml": "name: backup\nproject_name: infra-rustic\ncompose_files:\n - compose.yaml\n - compose.ops.yaml\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n profile: prod\n init_args:\n - --set-chunker\n - rabin\n - --set-chunk-size\n - 1MiB\n", }) bundleMux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(bundleTestServer{bundle: bundle, expectedTaskID: "task-rustic-init", responseServiceName: "backup", responseRelativeRoot: "backup"}, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) bundleMux.Handle(bundlePath, bundleHandler) bundleHTTPServer := httptest.NewServer(bundleMux) defer bundleHTTPServer.Close()
reportServer := &agentExecutionTestReportServer{} reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
bundleClient := agentv1connect.NewBundleServiceClient(bundleHTTPServer.Client(), bundleHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) logUploader := newTaskLogUploader(reportClient, "task-rustic-init") defer func() { _ = logUploader.Close() }()
pulledTask := &agentv1.AgentTask{TaskId: "task-rustic-init", Type: protoAgentTaskType(task.TypeRusticInit), ServiceName: "backup", NodeId: "main", ParamsJson: `{"service_dir":"backup"}`} if err := executeRusticInitTask(context.Background(), bundleClient, reportClient, cfg, pulledTask, logUploader); err != nil { t.Fatalf("execute rustic init task: %v", err) }
argsContent, err := os.ReadFile(argsFile) //nolint:gosec if err != nil { t.Fatalf("read args file: %v", err) } if got := string(argsContent); got != "compose --project-name infra-rustic -f compose.yaml -f compose.ops.yaml run --rm rustic -P prod init --set-chunker rabin --set-chunk-size 1MiB " { t.Fatalf("unexpected docker args %q", got) } pwdContent, err := os.ReadFile(pwdFile) //nolint:gosec if err != nil { t.Fatalf("read pwd file: %v", err) } if string(bytesTrimSpace(pwdContent)) != filepath.Join(cfg.RepoDir, "backup") { t.Fatalf("expected docker cwd %q, got %q", filepath.Join(cfg.RepoDir, "backup"), string(bytesTrimSpace(pwdContent))) } reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusSucceeded) { t.Fatalf("expected succeeded task status, got %q", reportServer.taskStatus) } if reportServer.stepStatuses[task.StepInit] != string(task.StatusSucceeded) { t.Fatalf("expected init step succeeded, got %+v", reportServer.stepStatuses) }}
func TestExecutePulledTaskWithTimeoutMarksTimedOutTaskFailed(t *testing.T) { t.Parallel()
reportServer := &agentExecutionTestReportServer{blockLogUploads: true} reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) pulledTask := &agentv1.AgentTask{TaskId: "task-timeout", ExecutionId: "execution-timeout", Type: protoAgentTaskType(task.TypePrune), NodeId: "main", ParamsJson: `{"target":"images_all"}`} statePath := taskExecutionStatePath(t.TempDir()) runtime := &taskExecutionRuntime{path: statePath, state: persistedTaskExecution{TaskID: pulledTask.GetTaskId(), ExecutionID: pulledTask.GetExecutionId(), Acknowledged: true}} if err := persistTaskExecution(statePath, runtime.state); err != nil { t.Fatal(err) } err := executePulledTaskWithTimeout(withTaskExecution(context.Background(), runtime), nil, reportClient, &config.AgentConfig{}, pulledTask, time.Second) if err == nil || !strings.Contains(err.Error(), "task exceeded execution timeout") { t.Fatalf("expected timeout error, got %v", err) }
reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusFailed) { t.Fatalf("expected failed task status, got %q", reportServer.taskStatus) } if reportServer.taskExecutionID != pulledTask.GetExecutionId() { t.Fatalf("expected execution_id %q, got %q", pulledTask.GetExecutionId(), reportServer.taskExecutionID) } if !strings.Contains(reportServer.taskErrorSummary, "task exceeded execution timeout") { t.Fatalf("expected timeout error summary, got %q", reportServer.taskErrorSummary) } if reportServer.confirmedSeq != 1 { t.Fatalf("expected initial task log upload attempt before timeout, got confirmed seq %d", reportServer.confirmedSeq) }}
func TestExecutePulledTaskWithTimeoutDoesNotFailForLogTransportErrors(t *testing.T) { installFakeDocker(t)
reportServer := &agentExecutionTestReportServer{wrongLogAckTaskID: true} reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) pulledTask := &agentv1.AgentTask{TaskId: "task-error", Type: protoAgentTaskType(task.TypePrune), NodeId: "main", ParamsJson: `{"target":"images_all"}`} err := executePulledTaskWithTimeout(context.Background(), nil, reportClient, &config.AgentConfig{}, pulledTask, time.Hour) if err != nil { t.Fatalf("expected log transport error to be non-fatal, got %v", err) }
reportServer.mu.Lock() defer reportServer.mu.Unlock() if reportServer.taskStatus != string(task.StatusSucceeded) { t.Fatalf("expected succeeded task status, got %q", reportServer.taskStatus) }}
func TestReportTaskCompletionWithTimeoutReturnsDeadlineExceeded(t *testing.T) { t.Parallel()
reportServer := &agentExecutionTestReportServer{blockTaskState: true} reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) err := reportTaskCompletionWithTimeout(context.Background(), reportClient, "task-timeout", task.StatusFailed, "boom", 50*time.Millisecond) if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("expected task completion timeout, got %v", err) }}
func TestReportTaskStepStateWithTimeoutReturnsDeadlineExceeded(t *testing.T) { t.Parallel()
reportServer := &agentExecutionTestReportServer{blockTaskStepState: true} reportMux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", errString("unexpected token") } return "main", nil }))) reportMux.Handle(reportPath, reportHandler) reportHTTPServer := httptest.NewUnstartedServer(reportMux) reportHTTPServer.EnableHTTP2 = true reportHTTPServer.StartTLS() defer reportHTTPServer.Close()
reportClient := agentv1connect.NewAgentReportServiceClient(reportHTTPServer.Client(), reportHTTPServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) err := reportTaskStepStateWithTimeout(context.Background(), reportClient, &agentv1.ReportTaskStepStateRequest{TaskId: "task-timeout", StepName: protoAgentTaskStepName(task.StepPrune), Status: protoAgentTaskStatus(task.StatusRunning)}, 50*time.Millisecond) if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("expected task step timeout, got %v", err) }}
type agentExecutionTestReportServer struct { agentv1connect.UnimplementedAgentReportServiceHandler mu sync.Mutex taskStatus string taskExecutionID string taskErrorSummary string runtimeStatus string stepStatuses map[task.StepName]string confirmedSeq uint64 backupArtifactRef string backupDataName string backupStatus string backupErrorSummary string blockLogUploads bool blockTaskState bool blockTaskStepState bool wrongLogAckTaskID bool}
func (server *agentExecutionTestReportServer) Heartbeat(context.Context, *connect.Request[agentv1.HeartbeatRequest]) (*connect.Response[agentv1.HeartbeatResponse], error) { return nil, connect.NewError(connect.CodeUnimplemented, errors.New("not used"))}
func (server *agentExecutionTestReportServer) ReportTaskState(ctx context.Context, req *connect.Request[agentv1.ReportTaskStateRequest]) (*connect.Response[agentv1.ReportTaskStateResponse], error) { if server.blockTaskState { <-ctx.Done() return nil, ctx.Err() } server.mu.Lock() defer server.mu.Unlock() server.taskStatus = agentTaskStatusText(req.Msg.GetStatus()) server.taskExecutionID = req.Msg.GetExecutionId() server.taskErrorSummary = req.Msg.GetErrorSummary() return connect.NewResponse(&agentv1.ReportTaskStateResponse{}), nil}
func (server *agentExecutionTestReportServer) ReportTaskStepState(ctx context.Context, req *connect.Request[agentv1.ReportTaskStepStateRequest]) (*connect.Response[agentv1.ReportTaskStepStateResponse], error) { if server.blockTaskStepState { <-ctx.Done() return nil, ctx.Err() } server.mu.Lock() defer server.mu.Unlock() if server.stepStatuses == nil { server.stepStatuses = make(map[task.StepName]string) } server.stepStatuses[agentTaskStepNameToTask(req.Msg.GetStepName())] = agentTaskStatusText(req.Msg.GetStatus()) return connect.NewResponse(&agentv1.ReportTaskStepStateResponse{}), nil}
func (server *agentExecutionTestReportServer) UploadTaskLogs(ctx context.Context, stream *connect.BidiStream[agentv1.UploadTaskLogsRequest, agentv1.UploadTaskLogsResponse]) error { for { message, err := stream.Receive() if err != nil { if errors.Is(err, io.EOF) { return nil } return err } server.mu.Lock() if message.GetSeq() > server.confirmedSeq { server.confirmedSeq = message.GetSeq() } confirmedSeq := server.confirmedSeq blockLogUploads := server.blockLogUploads wrongLogAckTaskID := server.wrongLogAckTaskID server.mu.Unlock() if blockLogUploads { <-ctx.Done() return ctx.Err() } ackTaskID := message.GetTaskId() if wrongLogAckTaskID { ackTaskID = "other-task" } if err := stream.Send(&agentv1.UploadTaskLogsResponse{TaskId: ackTaskID, LastConfirmedSeq: confirmedSeq}); err != nil { return err } }}
func (server *agentExecutionTestReportServer) ReportBackupResult(_ context.Context, req *connect.Request[agentv1.ReportBackupResultRequest]) (*connect.Response[agentv1.ReportBackupResultResponse], error) { server.mu.Lock() defer server.mu.Unlock() server.backupArtifactRef = req.Msg.GetArtifactRef() server.backupDataName = req.Msg.GetDataName() server.backupStatus = req.Msg.GetStatus() server.backupErrorSummary = req.Msg.GetErrorSummary() return connect.NewResponse(&agentv1.ReportBackupResultResponse{}), nil}
func (server *agentExecutionTestReportServer) ReportServiceInstanceStatus(_ context.Context, req *connect.Request[agentv1.ReportServiceInstanceStatusRequest]) (*connect.Response[agentv1.ReportServiceInstanceStatusResponse], error) { server.mu.Lock() defer server.mu.Unlock() server.runtimeStatus = req.Msg.GetRuntimeStatus() return connect.NewResponse(&agentv1.ReportServiceInstanceStatusResponse{}), nil}
var _ agentv1connect.AgentReportServiceHandler = (*agentExecutionTestReportServer)(nil)