diff --git a/api/callback.openapi.yaml b/api/callback.openapi.yaml index c89f988..c98d027 100644 --- a/api/callback.openapi.yaml +++ b/api/callback.openapi.yaml @@ -8,6 +8,28 @@ servers: tags: - name: worker paths: + /internal/v1/workers/{runtime_id}/progress: + post: + operationId: reportWorkerProgress + tags: [worker] + summary: Report progress for a pending worker action + parameters: + - $ref: '#/components/parameters/Runtime' + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/WorkerProgress' + responses: + '204': + description: Worker progress accepted + '400': + description: Invalid worker progress + '401': + description: Invalid workload identity + '409': + description: Worker action is no longer pending /internal/v1/workers/{runtime_id}/complete: post: operationId: completeWorker @@ -37,6 +59,15 @@ components: schema: type: string schemas: + WorkerProgress: + type: object + required: [percentage] + properties: + percentage: + type: integer + format: int64 + minimum: 0 + maximum: 100 WorkerResult: type: object properties: diff --git a/apps/druid/adapters/cli/callback.go b/apps/druid/adapters/cli/callback.go index 7027290..52950f6 100644 --- a/apps/druid/adapters/cli/callback.go +++ b/apps/druid/adapters/cli/callback.go @@ -12,6 +12,26 @@ type runtimeCallbackHandler struct { allowUnauthenticated bool } +func (h runtimeCallbackHandler) ReportWorkerProgress(c *fiber.Ctx, runtimeID callbackapi.Runtime) error { + var report callbackapi.WorkerProgress + if err := c.BodyParser(&report); err != nil { + return fiber.NewError(fiber.StatusBadRequest, "invalid progress report") + } + if report.Percentage < 0 || report.Percentage > 100 { + return fiber.NewError(fiber.StatusBadRequest, "percentage must be between 0 and 100") + } + if !h.allowUnauthenticated { + identity, ok := c.Locals("druid-workload-identity").(ports.RuntimeWorkloadIdentity) + if !ok || identity.Kind != "worker" || identity.RuntimeID != string(runtimeID) { + return fiber.NewError(fiber.StatusForbidden, "worker identity does not match runtime") + } + } + if err := h.callbacks.ReportProgress(string(runtimeID), report.Percentage); err != nil { + return fiber.NewError(fiber.StatusConflict, err.Error()) + } + return c.SendStatus(fiber.StatusNoContent) +} + func (h runtimeCallbackHandler) CompleteWorker(c *fiber.Ctx, runtimeID callbackapi.Runtime) error { var result callbackapi.WorkerResult if err := c.BodyParser(&result); err != nil { diff --git a/apps/druid/adapters/cli/callback_test.go b/apps/druid/adapters/cli/callback_test.go new file mode 100644 index 0000000..481fccd --- /dev/null +++ b/apps/druid/adapters/cli/callback_test.go @@ -0,0 +1,40 @@ +package cli + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gofiber/fiber/v2" + appservices "github.com/highcard-dev/daemon/apps/druid/core/services" + "github.com/highcard-dev/daemon/internal/callbackapi" +) + +func TestRuntimeCallbackHandlerReportsProgress(t *testing.T) { + callbacks := appservices.NewWorkerCallbackManager() + _, err := callbacks.Register("runtime-1") + if err != nil { + t.Fatal(err) + } + handler := runtimeCallbackHandler{callbacks: callbacks, allowUnauthenticated: true} + app := fiber.New() + callbackapi.RegisterHandlers(app, handler) + + request := httptest.NewRequest( + http.MethodPost, + "/internal/v1/workers/runtime-1/progress", + strings.NewReader(`{"percentage":42}`), + ) + request.Header.Set(fiber.HeaderContentType, fiber.MIMEApplicationJSON) + response, err := app.Test(request) + if err != nil { + t.Fatal(err) + } + if response.StatusCode != http.StatusNoContent { + t.Fatalf("status = %d; want %d", response.StatusCode, http.StatusNoContent) + } + if progress, ok := callbacks.Progress("runtime-1"); !ok || progress != 42 { + t.Fatalf("progress = %v, %v; want 42, true", progress, ok) + } +} diff --git a/apps/druid/adapters/cli/daemon.go b/apps/druid/adapters/cli/daemon.go index 7209c1a..8dda027 100644 --- a/apps/druid/adapters/cli/daemon.go +++ b/apps/druid/adapters/cli/daemon.go @@ -190,7 +190,7 @@ func runRuntimeDaemon() error { websocketHandler.SetAllowUnauthenticatedPublic(runtimeAllowUnauthenticatedPublic) handlers := runtimehandlers.RouteHandlers{ Server: runtimehandlers.NewRuntimeServer( - runtimehandlers.NewHealthHandler(), + runtimehandlers.NewHealthHandlerWithProgress(callbacks.Progress), scrollHandler, ), Websocket: websocketHandler, diff --git a/apps/druid/adapters/cli/worker_progress_test.go b/apps/druid/adapters/cli/worker_progress_test.go new file mode 100644 index 0000000..7b5add8 --- /dev/null +++ b/apps/druid/adapters/cli/worker_progress_test.go @@ -0,0 +1,58 @@ +package cli + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "github.com/highcard-dev/daemon/internal/core/domain" + "github.com/highcard-dev/daemon/internal/core/ports" +) + +func TestWorkerProgressReporterReadsSnapshotProgress(t *testing.T) { + reports := make(chan int64, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + var report struct { + Percentage int64 `json:"percentage"` + } + if err := json.NewDecoder(request.Body).Decode(&report); err != nil { + t.Error(err) + } + if got := request.Header.Get("Authorization"); got != "Bearer worker-token" { + t.Errorf("authorization = %q; want Bearer worker-token", got) + } + reports <- report.Percentage + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + progress := domain.NewSnapshotProgress() + progress.Percentage.Store(37) + tokenFile := filepath.Join(t.TempDir(), "token") + if err := os.WriteFile(tokenFile, []byte("worker-token\n"), 0600); err != nil { + t.Fatal(err) + } + stop := startWorkerProgressReporter( + ports.RuntimeWorkerAction{ + RuntimeID: "runtime-1", + CallbackURL: server.URL + "/internal/v1/workers/runtime-1/complete", + TokenFile: tokenFile, + }, + progress, + time.Hour, + ) + defer stop() + + select { + case percentage := <-reports: + if percentage != 37 { + t.Fatalf("percentage = %d; want 37", percentage) + } + case <-time.After(time.Second): + t.Fatal("progress was not reported") + } +} diff --git a/apps/druid/adapters/cli/worker_pull.go b/apps/druid/adapters/cli/worker_pull.go index 4bdab15..17726c3 100644 --- a/apps/druid/adapters/cli/worker_pull.go +++ b/apps/druid/adapters/cli/worker_pull.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "github.com/highcard-dev/daemon/internal/callbackapi" @@ -66,6 +67,10 @@ func runWorkerPull(action ports.RuntimeWorkerAction) ports.RuntimeWorkerResult { if root == "" { root = "/scroll" } + progress := domain.NewSnapshotProgress() + stopProgress := startWorkerProgressReporter(action, progress, time.Second) + defer stopProgress() + oci := registry.NewOciClient(loadWorkerRegistryStore()) digest, err := oci.ResolveDigest(action.Artifact) if err == nil { @@ -73,11 +78,11 @@ func runWorkerPull(action ports.RuntimeWorkerAction) ports.RuntimeWorkerResult { } switch action.Mode { case ports.RuntimeWorkerModeUpdate: - err = pullWorkerUpdate(root, action.Artifact, oci) + err = pullWorkerUpdate(root, action.Artifact, oci, progress) case ports.RuntimeWorkerModeRestore: - err = pullWorkerRestore(root, action.Artifact, oci) + err = pullWorkerRestore(root, action.Artifact, oci, progress) default: - err = pullWorkerCreate(root, action.Artifact, oci) + err = pullWorkerCreate(root, action.Artifact, oci, progress) } if err != nil { result.Error = err.Error() @@ -116,7 +121,7 @@ func loadWorkerRegistryStore() *registry.CredentialStore { return registry.NewCredentialStore(config.Registries) } -func pullWorkerCreate(root string, artifact string, oci ports.OciRegistryInterface) error { +func pullWorkerCreate(root string, artifact string, oci ports.OciRegistryInterface, progress *domain.SnapshotProgress) error { if err := os.MkdirAll(root, 0755); err != nil { return err } @@ -138,16 +143,16 @@ func pullWorkerCreate(root string, artifact string, oci ports.OciRegistryInterfa } return copyPath(artifact, root) } - return oci.PullSelective(root, artifact, true, nil) + return oci.PullSelective(root, artifact, true, progress) } -func pullWorkerUpdate(root string, artifact string, oci ports.OciRegistryInterface) error { +func pullWorkerUpdate(root string, artifact string, oci ports.OciRegistryInterface, progress *domain.SnapshotProgress) error { tmp, err := os.MkdirTemp("", "druid-worker-update-*") if err != nil { return err } defer os.RemoveAll(tmp) - if err := coreservices.MaterializeScrollArtifact(artifact, tmp, oci, true); err != nil { + if err := coreservices.MaterializeScrollArtifactWithProgress(artifact, tmp, oci, true, progress); err != nil { return err } scrollYAML, err := os.ReadFile(filepath.Join(tmp, "scroll.yaml")) @@ -163,13 +168,13 @@ func pullWorkerUpdate(root string, artifact string, oci ports.OciRegistryInterfa return mergePulledRoot(tmp, root, skipData) } -func pullWorkerRestore(root string, artifact string, oci ports.OciRegistryInterface) error { +func pullWorkerRestore(root string, artifact string, oci ports.OciRegistryInterface, progress *domain.SnapshotProgress) error { tmp, err := os.MkdirTemp("", "druid-worker-restore-*") if err != nil { return err } defer os.RemoveAll(tmp) - if err := coreservices.MaterializeScrollArtifact(artifact, tmp, oci, true); err != nil { + if err := coreservices.MaterializeScrollArtifactWithProgress(artifact, tmp, oci, true, progress); err != nil { return err } if err := os.MkdirAll(root, 0755); err != nil { @@ -314,6 +319,69 @@ func copyPath(src string, dst string) error { return err } +func startWorkerProgressReporter(action ports.RuntimeWorkerAction, progress *domain.SnapshotProgress, interval time.Duration) func() { + if action.CallbackURL == "" || action.TokenFile == "" || progress == nil { + return func() {} + } + suffix := "/internal/v1/workers/" + action.RuntimeID + "/complete" + baseURL := strings.TrimSuffix(action.CallbackURL, suffix) + if baseURL == action.CallbackURL { + return func() {} + } + client, err := callbackapi.NewClientWithResponses(baseURL) + if err != nil { + return func() {} + } + done := make(chan struct{}) + var wait sync.WaitGroup + wait.Add(1) + go func() { + defer wait.Done() + ticker := time.NewTicker(interval) + defer ticker.Stop() + lastPercentage := int64(-1) + report := func() { + percentage := progress.Percentage.Load() + if percentage == lastPercentage { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + response, err := client.ReportWorkerProgressWithResponse( + ctx, + action.RuntimeID, + callbackapi.WorkerProgress{Percentage: percentage}, + func(_ context.Context, request *http.Request) error { + token, err := os.ReadFile(action.TokenFile) + if err != nil { + return fmt.Errorf("read worker token: %w", err) + } + request.Header.Set("Authorization", "Bearer "+strings.TrimSpace(string(token))) + return nil + }, + ) + cancel() + if err == nil && response.StatusCode() < http.StatusBadRequest { + lastPercentage = percentage + } + } + report() + for { + select { + case <-ticker.C: + report() + case <-done: + report() + return + } + } + }() + var once sync.Once + return func() { + once.Do(func() { close(done) }) + wait.Wait() + } +} + func reportWorkerResult(action ports.RuntimeWorkerAction, result ports.RuntimeWorkerResult) error { if action.CallbackURL == "" { body, err := json.Marshal(result) diff --git a/apps/druid/adapters/cli/worker_test.go b/apps/druid/adapters/cli/worker_test.go index 80a3cb2..509c113 100644 --- a/apps/druid/adapters/cli/worker_test.go +++ b/apps/druid/adapters/cli/worker_test.go @@ -97,7 +97,7 @@ func TestWorkerRestoreStagesBeforeReplacingRoot(t *testing.T) { mustWrite(t, filepath.Join(root, "data", "old-only.txt"), "old") oci := fakeRestoreOCI{t: t} - if err := pullWorkerRestore(root, "registry.local/backup:1", oci); err != nil { + if err := pullWorkerRestore(root, "registry.local/backup:1", oci, nil); err != nil { t.Fatal(err) } diff --git a/apps/druid/adapters/http/handlers/health_handler.go b/apps/druid/adapters/http/handlers/health_handler.go index 0079f5e..e7658d0 100644 --- a/apps/druid/adapters/http/handlers/health_handler.go +++ b/apps/druid/adapters/http/handlers/health_handler.go @@ -5,12 +5,27 @@ import ( "github.com/highcard-dev/daemon/internal/api" ) -type HealthHandler struct{} +type ProgressLookup func(runtimeID string) (float64, bool) + +type HealthHandler struct { + progress ProgressLookup +} func NewHealthHandler() *HealthHandler { return &HealthHandler{} } +func NewHealthHandlerWithProgress(progress ProgressLookup) *HealthHandler { + return &HealthHandler{progress: progress} +} + func (h *HealthHandler) GetHealthAuth(c *fiber.Ctx) error { - return c.JSON(api.HealthResponse{Mode: "ok"}) + health := api.HealthResponse{Mode: "ok"} + if h.progress != nil { + if progress, ok := h.progress(c.Params("id")); ok { + value := float32(progress) + health.Progress = &value + } + } + return c.JSON(health) } diff --git a/apps/druid/adapters/http/handlers/health_handler_test.go b/apps/druid/adapters/http/handlers/health_handler_test.go new file mode 100644 index 0000000..8678e6b --- /dev/null +++ b/apps/druid/adapters/http/handlers/health_handler_test.go @@ -0,0 +1,33 @@ +package handlers + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gofiber/fiber/v2" + "github.com/highcard-dev/daemon/internal/api" +) + +func TestGetHealthAuthIncludesPullProgress(t *testing.T) { + handler := NewHealthHandlerWithProgress(func(runtimeID string) (float64, bool) { + return 37, runtimeID == "scroll-1" + }) + app := fiber.New() + app.Get("/:id/api/v1/health", handler.GetHealthAuth) + + response, err := app.Test(httptest.NewRequest(http.MethodGet, "/scroll-1/api/v1/health", nil)) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + + var health api.HealthResponse + if err := json.NewDecoder(response.Body).Decode(&health); err != nil { + t.Fatal(err) + } + if health.Progress == nil || *health.Progress != 37 { + t.Fatalf("progress = %v; want 37", health.Progress) + } +} diff --git a/apps/druid/core/services/worker_callbacks.go b/apps/druid/core/services/worker_callbacks.go index bd8c999..00d1c91 100644 --- a/apps/druid/core/services/worker_callbacks.go +++ b/apps/druid/core/services/worker_callbacks.go @@ -8,8 +8,9 @@ import ( ) type WorkerCallbackManager struct { - mu sync.Mutex - actions map[string]workerCallbackAction + mu sync.Mutex + actions map[string]workerCallbackAction + progress map[string]int64 } type workerCallbackAction struct { @@ -17,7 +18,10 @@ type workerCallbackAction struct { } func NewWorkerCallbackManager() *WorkerCallbackManager { - return &WorkerCallbackManager{actions: map[string]workerCallbackAction{}} + return &WorkerCallbackManager{ + actions: map[string]workerCallbackAction{}, + progress: map[string]int64{}, + } } func (m *WorkerCallbackManager) Register(runtimeID string) (<-chan ports.RuntimeWorkerResult, error) { @@ -28,6 +32,7 @@ func (m *WorkerCallbackManager) Register(runtimeID string) (<-chan ports.Runtime return nil, fmt.Errorf("worker action already pending for runtime %s", runtimeID) } m.actions[runtimeID] = workerCallbackAction{result: ch} + m.progress[runtimeID] = 0 m.mu.Unlock() return ch, nil } @@ -35,9 +40,27 @@ func (m *WorkerCallbackManager) Register(runtimeID string) (<-chan ports.Runtime func (m *WorkerCallbackManager) Cancel(runtimeID string) { m.mu.Lock() delete(m.actions, runtimeID) + delete(m.progress, runtimeID) m.mu.Unlock() } +func (m *WorkerCallbackManager) ReportProgress(runtimeID string, percentage int64) error { + m.mu.Lock() + defer m.mu.Unlock() + if _, ok := m.actions[runtimeID]; !ok { + return fmt.Errorf("unknown or completed worker action") + } + m.progress[runtimeID] = max(0, min(100, percentage)) + return nil +} + +func (m *WorkerCallbackManager) Progress(runtimeID string) (float64, bool) { + m.mu.Lock() + defer m.mu.Unlock() + progress, ok := m.progress[runtimeID] + return float64(progress), ok +} + func (m *WorkerCallbackManager) Complete(runtimeID string, result ports.RuntimeWorkerResult) error { m.mu.Lock() action, ok := m.actions[runtimeID] @@ -46,6 +69,7 @@ func (m *WorkerCallbackManager) Complete(runtimeID string, result ports.RuntimeW return fmt.Errorf("unknown or completed worker action") } delete(m.actions, runtimeID) + delete(m.progress, runtimeID) m.mu.Unlock() action.result <- result close(action.result) diff --git a/apps/druid/core/services/worker_callbacks_test.go b/apps/druid/core/services/worker_callbacks_test.go index 81dff22..2917456 100644 --- a/apps/druid/core/services/worker_callbacks_test.go +++ b/apps/druid/core/services/worker_callbacks_test.go @@ -50,3 +50,33 @@ func TestWorkerCallbackRejectsUnknownRuntime(t *testing.T) { t.Fatal("unknown runtime should fail") } } + +func TestWorkerCallbackTracksPullProgress(t *testing.T) { + manager := NewWorkerCallbackManager() + _, err := manager.Register("scroll-a") + if err != nil { + t.Fatal(err) + } + + if progress, ok := manager.Progress("scroll-a"); !ok || progress != 0 { + t.Fatalf("initial progress = %v, %v; want 0, true", progress, ok) + } + if err := manager.ReportProgress("scroll-a", 42); err != nil { + t.Fatal(err) + } + if progress, ok := manager.Progress("scroll-a"); !ok || progress != 42 { + t.Fatalf("reported progress = %v, %v; want 42, true", progress, ok) + } + + manager.Cancel("scroll-a") + if _, ok := manager.Progress("scroll-a"); ok { + t.Fatal("cancelled progress should be removed") + } +} + +func TestWorkerCallbackRejectsProgressForUnknownRuntime(t *testing.T) { + manager := NewWorkerCallbackManager() + if err := manager.ReportProgress("missing", 42); err == nil { + t.Fatal("progress for an unknown worker action should fail") + } +} diff --git a/internal/callbackapi/generated.go b/internal/callbackapi/generated.go index 95a77a4..d147979 100644 --- a/internal/callbackapi/generated.go +++ b/internal/callbackapi/generated.go @@ -21,6 +21,11 @@ import ( "github.com/oapi-codegen/runtime" ) +// WorkerProgress defines model for WorkerProgress. +type WorkerProgress struct { + Percentage int64 `json:"percentage"` +} + // WorkerResult defines model for WorkerResult. type WorkerResult struct { ArtifactDigest *string `json:"artifact_digest,omitempty"` @@ -34,6 +39,9 @@ type Runtime = string // CompleteWorkerJSONRequestBody defines body for CompleteWorker for application/json ContentType. type CompleteWorkerJSONRequestBody = WorkerResult +// ReportWorkerProgressJSONRequestBody defines body for ReportWorkerProgress for application/json ContentType. +type ReportWorkerProgressJSONRequestBody = WorkerProgress + // RequestEditorFn is the function signature for the RequestEditor callback function type RequestEditorFn func(ctx context.Context, req *http.Request) error @@ -111,6 +119,11 @@ type ClientInterface interface { CompleteWorkerWithBody(ctx context.Context, runtimeId Runtime, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) CompleteWorker(ctx context.Context, runtimeId Runtime, body CompleteWorkerJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) + + // ReportWorkerProgressWithBody request with any body + ReportWorkerProgressWithBody(ctx context.Context, runtimeId Runtime, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) + + ReportWorkerProgress(ctx context.Context, runtimeId Runtime, body ReportWorkerProgressJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) } func (c *Client) CompleteWorkerWithBody(ctx context.Context, runtimeId Runtime, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) { @@ -137,6 +150,30 @@ func (c *Client) CompleteWorker(ctx context.Context, runtimeId Runtime, body Com return c.Client.Do(req) } +func (c *Client) ReportWorkerProgressWithBody(ctx context.Context, runtimeId Runtime, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) { + req, err := NewReportWorkerProgressRequestWithBody(c.Server, runtimeId, contentType, body) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + if err := c.applyEditors(ctx, req, reqEditors); err != nil { + return nil, err + } + return c.Client.Do(req) +} + +func (c *Client) ReportWorkerProgress(ctx context.Context, runtimeId Runtime, body ReportWorkerProgressJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) { + req, err := NewReportWorkerProgressRequest(c.Server, runtimeId, body) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + if err := c.applyEditors(ctx, req, reqEditors); err != nil { + return nil, err + } + return c.Client.Do(req) +} + // NewCompleteWorkerRequest calls the generic CompleteWorker builder with application/json body func NewCompleteWorkerRequest(server string, runtimeId Runtime, body CompleteWorkerJSONRequestBody) (*http.Request, error) { var bodyReader io.Reader @@ -184,6 +221,53 @@ func NewCompleteWorkerRequestWithBody(server string, runtimeId Runtime, contentT return req, nil } +// NewReportWorkerProgressRequest calls the generic ReportWorkerProgress builder with application/json body +func NewReportWorkerProgressRequest(server string, runtimeId Runtime, body ReportWorkerProgressJSONRequestBody) (*http.Request, error) { + var bodyReader io.Reader + buf, err := json.Marshal(body) + if err != nil { + return nil, err + } + bodyReader = bytes.NewReader(buf) + return NewReportWorkerProgressRequestWithBody(server, runtimeId, "application/json", bodyReader) +} + +// NewReportWorkerProgressRequestWithBody generates requests for ReportWorkerProgress with any type of body +func NewReportWorkerProgressRequestWithBody(server string, runtimeId Runtime, contentType string, body io.Reader) (*http.Request, error) { + var err error + + var pathParam0 string + + pathParam0, err = runtime.StyleParamWithLocation("simple", false, "runtime_id", runtime.ParamLocationPath, runtimeId) + if err != nil { + return nil, err + } + + serverURL, err := url.Parse(server) + if err != nil { + return nil, err + } + + operationPath := fmt.Sprintf("/internal/v1/workers/%s/progress", pathParam0) + if operationPath[0] == '/' { + operationPath = "." + operationPath + } + + queryURL, err := serverURL.Parse(operationPath) + if err != nil { + return nil, err + } + + req, err := http.NewRequest("POST", queryURL.String(), body) + if err != nil { + return nil, err + } + + req.Header.Add("Content-Type", contentType) + + return req, nil +} + func (c *Client) applyEditors(ctx context.Context, req *http.Request, additionalEditors []RequestEditorFn) error { for _, r := range c.RequestEditors { if err := r(ctx, req); err != nil { @@ -231,6 +315,11 @@ type ClientWithResponsesInterface interface { CompleteWorkerWithBodyWithResponse(ctx context.Context, runtimeId Runtime, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*CompleteWorkerResponse, error) CompleteWorkerWithResponse(ctx context.Context, runtimeId Runtime, body CompleteWorkerJSONRequestBody, reqEditors ...RequestEditorFn) (*CompleteWorkerResponse, error) + + // ReportWorkerProgressWithBodyWithResponse request with any body + ReportWorkerProgressWithBodyWithResponse(ctx context.Context, runtimeId Runtime, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*ReportWorkerProgressResponse, error) + + ReportWorkerProgressWithResponse(ctx context.Context, runtimeId Runtime, body ReportWorkerProgressJSONRequestBody, reqEditors ...RequestEditorFn) (*ReportWorkerProgressResponse, error) } type CompleteWorkerResponse struct { @@ -254,6 +343,27 @@ func (r CompleteWorkerResponse) StatusCode() int { return 0 } +type ReportWorkerProgressResponse struct { + Body []byte + HTTPResponse *http.Response +} + +// Status returns HTTPResponse.Status +func (r ReportWorkerProgressResponse) Status() string { + if r.HTTPResponse != nil { + return r.HTTPResponse.Status + } + return http.StatusText(0) +} + +// StatusCode returns HTTPResponse.StatusCode +func (r ReportWorkerProgressResponse) StatusCode() int { + if r.HTTPResponse != nil { + return r.HTTPResponse.StatusCode + } + return 0 +} + // CompleteWorkerWithBodyWithResponse request with arbitrary body returning *CompleteWorkerResponse func (c *ClientWithResponses) CompleteWorkerWithBodyWithResponse(ctx context.Context, runtimeId Runtime, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*CompleteWorkerResponse, error) { rsp, err := c.CompleteWorkerWithBody(ctx, runtimeId, contentType, body, reqEditors...) @@ -271,6 +381,23 @@ func (c *ClientWithResponses) CompleteWorkerWithResponse(ctx context.Context, ru return ParseCompleteWorkerResponse(rsp) } +// ReportWorkerProgressWithBodyWithResponse request with arbitrary body returning *ReportWorkerProgressResponse +func (c *ClientWithResponses) ReportWorkerProgressWithBodyWithResponse(ctx context.Context, runtimeId Runtime, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*ReportWorkerProgressResponse, error) { + rsp, err := c.ReportWorkerProgressWithBody(ctx, runtimeId, contentType, body, reqEditors...) + if err != nil { + return nil, err + } + return ParseReportWorkerProgressResponse(rsp) +} + +func (c *ClientWithResponses) ReportWorkerProgressWithResponse(ctx context.Context, runtimeId Runtime, body ReportWorkerProgressJSONRequestBody, reqEditors ...RequestEditorFn) (*ReportWorkerProgressResponse, error) { + rsp, err := c.ReportWorkerProgress(ctx, runtimeId, body, reqEditors...) + if err != nil { + return nil, err + } + return ParseReportWorkerProgressResponse(rsp) +} + // ParseCompleteWorkerResponse parses an HTTP response from a CompleteWorkerWithResponse call func ParseCompleteWorkerResponse(rsp *http.Response) (*CompleteWorkerResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) @@ -287,11 +414,30 @@ func ParseCompleteWorkerResponse(rsp *http.Response) (*CompleteWorkerResponse, e return response, nil } +// ParseReportWorkerProgressResponse parses an HTTP response from a ReportWorkerProgressWithResponse call +func ParseReportWorkerProgressResponse(rsp *http.Response) (*ReportWorkerProgressResponse, error) { + bodyBytes, err := io.ReadAll(rsp.Body) + defer func() { _ = rsp.Body.Close() }() + if err != nil { + return nil, err + } + + response := &ReportWorkerProgressResponse{ + Body: bodyBytes, + HTTPResponse: rsp, + } + + return response, nil +} + // ServerInterface represents all server handlers. type ServerInterface interface { // Complete a pending worker action // (POST /internal/v1/workers/{runtime_id}/complete) CompleteWorker(c *fiber.Ctx, runtimeId Runtime) error + // Report progress for a pending worker action + // (POST /internal/v1/workers/{runtime_id}/progress) + ReportWorkerProgress(c *fiber.Ctx, runtimeId Runtime) error } // ServerInterfaceWrapper converts contexts to parameters. @@ -317,6 +463,22 @@ func (siw *ServerInterfaceWrapper) CompleteWorker(c *fiber.Ctx) error { return siw.Handler.CompleteWorker(c, runtimeId) } +// ReportWorkerProgress operation middleware +func (siw *ServerInterfaceWrapper) ReportWorkerProgress(c *fiber.Ctx) error { + + var err error + + // ------------- Path parameter "runtime_id" ------------- + var runtimeId Runtime + + err = runtime.BindStyledParameterWithOptions("simple", "runtime_id", c.Params("runtime_id"), &runtimeId, runtime.BindStyledParameterOptions{Explode: false, Required: true}) + if err != nil { + return fiber.NewError(fiber.StatusBadRequest, fmt.Errorf("Invalid format for parameter runtime_id: %w", err).Error()) + } + + return siw.Handler.ReportWorkerProgress(c, runtimeId) +} + // FiberServerOptions provides options for the Fiber server. type FiberServerOptions struct { BaseURL string @@ -340,19 +502,24 @@ func RegisterHandlersWithOptions(router fiber.Router, si ServerInterface, option router.Post(options.BaseURL+"/internal/v1/workers/:runtime_id/complete", wrapper.CompleteWorker) + router.Post(options.BaseURL+"/internal/v1/workers/:runtime_id/progress", wrapper.ReportWorkerProgress) + } // Base64 encoded, gzipped, json marshaled Swagger object var swaggerSpec = []string{ - "H4sIAAAAAAAC/3xTsW7bMBD9FeLaUQidNpO2Nl2yFVk6FIZxJs8OE4pkjycXgqB/L0jKtYMa3ezj43u8", - "955mMHFIMVCQDP0MCRkHEuL673kM4gYqP12AHhLKC3QQsMyA2+nOWeiA6dfomCz0wiN1kM0LDVhuypQK", - "Ogu7cIRlWc6HVeJH5DfiZ8qjl/oAjolYHNVTZHEHNLKz7khZbtB1QMyRb55kw9H73YSDv/WQ7jyJ+1cy", - "0p7mwiEWsKVs2CVxsSz+FIQ4oFcGvd+jeVNfvj+pMZNV+0n9rjtkhcEqSyeViU/E+Q46ECe+SHzj0Vn1", - "l+bxigY6KOgmtLm7v9uUt8dEAZODHj7XUVfNr6Zot9Lo071etfV8SWPRJVNPUoNLsflWbMWyzpOFHh5X", - "RLO/sl+C/znDR6YD9PBBX+qhLxB9LsaybclTlq/RTkXHxCAUqiSm5J2povo1l/3mq17ckliLod+1ouby", - "vl91kFMMufXk0+bh39AaieLKotAYSkK2mPuw2dzK+ITe2TXN9VpD3/8f7SNa5SwFcTLVYuVxGJCnK6MV", - "qkTBunA8C6CpVB0IHovp0OawbZ9I61ANY2QPPejq9gqezx/hemnZLn8CAAD//+gup1vLAwAA", + "H4sIAAAAAAAC/8xVTW8TMRD9K9bAcdXdQIXE3qBceqt64VBFlWtPtm79xXg2EEX735HtzUfbACpcuNXj", + "8Xsz771ttqCCi8Gj5wT9FqIk6ZCRyul69Gwc5j+Nhx6i5HtowMtcA6q3t0ZDA4TfRkOooWcasYGk7tHJ", + "/JI3MXcnJuMHmKZpd1kovgZ6RLqiMBCmOgKFiMQG6wlJoWc5lClWgZxk6MF4/nAODTj5w7jRQb/ougac", + "8fXUNTta4xkHJMi0hxlvjnGX++Zw94CKYWrmsa4xjZZfDiWJzUoqvtVmwMQntmwAiQKdvEmKgrW3G+ns", + "KX1eTJNLxq9CbtaYFJnIJmQ/Lj0jeWmFktbeSfUoPl1dijGhFncb8b3skIT0Wmhci4S0Rkpn0AAbtpni", + "C41Giz3MxREMNJC7K1F3tjjr8uwhopfRQA/vS6kpmSiitGaGadeLduZut4eQTG2OmkUuTsZQdcuyyrzO", + "pYYeLuaOKn9BP+TxZgtvCVfQw5v2kNr20NLu8jotq9mY+HPQm8yjgmf0hVLGaI0qpO1Dyvttj+J6imLO", + "a/skFdPTSOXYl0KKwaeak3fd+UvTKoiggiKkUhgZdRb3vOtOebyW1ujZzflZ7V78vtsGqYXR6NnwpgQr", + "jc5J2hwJLaSI6LXxw45AqgLVAMshiw61DssM8GeP4/GXfNLja4yB+Nl3/386vR/vH73eifJqt/dq/oXf", + "+cXHX45UbRYmCR+EDX7IbDUJz6JS/TrssAr0qtRksPqfpxg7koUe2uLc3Lzd/aLMj6bl9DMAAP//GC/e", + "GZgGAAA=", } // GetSwagger returns the content of the embedded swagger specification file diff --git a/internal/core/services/registry/oci.go b/internal/core/services/registry/oci.go index 9d9dbdc..226ebfd 100644 --- a/internal/core/services/registry/oci.go +++ b/internal/core/services/registry/oci.go @@ -201,6 +201,13 @@ func (c *OciClient) PullSelective(dir string, artifact string, includeData bool, if progress != nil { progress.Mode.Store(domain.SnapshotProgressModeRestore) progress.Percentage.Store(0) + defer progress.Mode.Store(domain.SnapshotProgressModeIdle) + } + storeProgress := func(done, total int64) { + if progress == nil || total <= 0 { + return + } + progress.Percentage.Store(min(99, done*100/total)) } copyOpts := oras.CopyOptions{ @@ -259,10 +266,7 @@ func (c *OciClient) PullSelective(dir string, artifact string, includeData bool, done := completed.Add(1) total := totalLayers.Load() bytesDownloaded.Add(desc.Size) - if progress != nil && total > 0 { - pct := done * 100 / total - progress.Percentage.Store(pct) - } + storeProgress(done, total) title := desc.Annotations["org.opencontainers.image.title"] logger.Log().Debug("Pulled layer", zap.String("title", title), @@ -279,10 +283,7 @@ func (c *OciClient) PullSelective(dir string, artifact string, includeData bool, done := completed.Add(1) total := totalLayers.Load() bytesDownloaded.Add(desc.Size) - if progress != nil && total > 0 { - pct := done * 100 / total - progress.Percentage.Store(pct) - } + storeProgress(done, total) title := desc.Annotations["org.opencontainers.image.title"] logger.Log().Debug("Layer already exists locally, skipped", zap.String("title", title), @@ -303,17 +304,9 @@ func (c *OciClient) PullSelective(dir string, artifact string, includeData bool, manifestDescriptor, err := oras.Copy(ctx, repoInstance, ref, fs, dstRef, copyOpts) stopProgress() if err != nil { - if progress != nil { - progress.Mode.Store(domain.SnapshotProgressModeIdle) - } return err } - if progress != nil { - progress.Percentage.Store(100) - progress.Mode.Store(domain.SnapshotProgressModeIdle) - } - logger.Log().Info("Manifest pulled", zap.String("digest", manifestDescriptor.Digest.String()), zap.String("mediaType", manifestDescriptor.MediaType)) jsonData, err := json.Marshal(&manifestDescriptor) @@ -349,6 +342,10 @@ func (c *OciClient) PullSelective(dir string, artifact string, includeData bool, return fmt.Errorf("failed to write annotations: %w", err) } + if progress != nil { + progress.Percentage.Store(100) + } + return nil } diff --git a/internal/core/services/registry/oci_test.go b/internal/core/services/registry/oci_test.go index 2179b3d..e8d5a84 100644 --- a/internal/core/services/registry/oci_test.go +++ b/internal/core/services/registry/oci_test.go @@ -117,6 +117,47 @@ func fakeRegistry(t *testing.T) *httptest.Server { return srv } +func TestPullSelectiveDoesNotReportCompleteBeforeMetadataIsWritten(t *testing.T) { + t.Chdir(t.TempDir()) + server := fakeRegistry(t) + registryHost := strings.TrimPrefix(server.URL, "http://") + source := filepath.Join("scrolls", "progress-test") + if err := os.MkdirAll(source, 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile( + filepath.Join(source, "scroll.yaml"), + []byte("name: progress-test\nversion: 0.1.0\napp_version: \"1.0\"\n"), + 0644, + ); err != nil { + t.Fatal(err) + } + + client := &OciClient{ + credentialStore: NewCredentialStore(nil), + plainHTTP: true, + } + repository := registryHost + "/test/progress" + if _, err := client.Push(source, repository, "1.0", nil, false, nil); err != nil { + t.Fatal(err) + } + + destination := filepath.Join("pull", "progress-test") + if err := os.MkdirAll(filepath.Join(destination, "manifest.json"), 0755); err != nil { + t.Fatal(err) + } + progress := domain.NewSnapshotProgress() + if err := client.PullSelective(destination, repository+":1.0", true, progress); err == nil { + t.Fatal("pull should fail when manifest.json cannot be written") + } + if percentage := progress.Percentage.Load(); percentage >= 100 { + t.Fatalf("failed pull progress = %d; want below 100", percentage) + } + if mode := progress.Mode.Load(); mode != domain.SnapshotProgressModeIdle { + t.Fatalf("failed pull mode = %v; want idle", mode) + } +} + func TestValidateCredentialsUsesPlainHTTPEnv(t *testing.T) { t.Setenv("DRUID_REGISTRY_PLAIN_HTTP", "true") diff --git a/internal/core/services/runtime_scroll_manager.go b/internal/core/services/runtime_scroll_manager.go index a431cb1..ce0aba5 100644 --- a/internal/core/services/runtime_scroll_manager.go +++ b/internal/core/services/runtime_scroll_manager.go @@ -102,6 +102,10 @@ func RuntimeScrollIDFromName(name string) string { } func MaterializeScrollArtifact(artifact string, root string, ociRegistry ports.OciRegistryInterface, includeData bool) error { + return MaterializeScrollArtifactWithProgress(artifact, root, ociRegistry, includeData, nil) +} + +func MaterializeScrollArtifactWithProgress(artifact string, root string, ociRegistry ports.OciRegistryInterface, includeData bool, progress *domain.SnapshotProgress) error { if artifact == "" { return fmt.Errorf("artifact is required") } @@ -126,7 +130,7 @@ func MaterializeScrollArtifact(artifact string, root string, ociRegistry ports.O if ociRegistry == nil { return fmt.Errorf("OCI registry is required to pull %s", artifact) } - if err := ociRegistry.PullSelective(root, artifact, includeData, nil); err != nil { + if err := ociRegistry.PullSelective(root, artifact, includeData, progress); err != nil { return err } return os.MkdirAll(filepath.Join(root, domain.RuntimeDataDir), 0755)