From e3351571c7ea639b233e06f20e05b3522924f064 Mon Sep 17 00:00:00 2001 From: utkuerol Date: Fri, 17 Jul 2026 19:17:54 +0200 Subject: [PATCH 1/4] update backup model --- model/backups.go | 64 +++++++++++++++++++++++++++++++++++++++++++++ model/common.go | 57 ---------------------------------------- model/kubernetes.go | 4 +++ 3 files changed, 68 insertions(+), 57 deletions(-) create mode 100644 model/backups.go diff --git a/model/backups.go b/model/backups.go new file mode 100644 index 0000000..cc97e94 --- /dev/null +++ b/model/backups.go @@ -0,0 +1,64 @@ +// Copyright (c) Codesphere Inc. +// SPDX-License-Identifier: Apache-2.0 + +package model + +// BackupStoreConfig holds configuration for backup storage. +type BackupStoreConfig struct { + // EndpointURL is the S3-compatible endpoint URL. + EndpointURL string `json:"endpointUrl"` + + // DestinationPath is the path/bucket for backups. + DestinationPath string `json:"destinationPath"` + + // AccessKeyID is the access key ID for the backup store. + AccessKeyID string `json:"accessKeyId"` +} + +// BackupStoreSecrets holds sensitive data for backup storage. +type BackupStoreSecrets struct { + // SecretKey is the secret access key. + SecretKey string `json:"secretKey"` +} + +// RecoverFromBackup specifies recovery from a specific backup. +type RecoverFromBackup struct { + // ID is the backup ID to recover from. + ID string `json:"id"` + + // Config is the backup store configuration. + Config BackupStoreConfig `json:"config"` + + // Secrets is the backup store secrets. + Secrets BackupStoreSecrets `json:"secrets"` +} + +// RecoverFromTimestamp specifies point-in-time recovery. +type RecoverFromTimestamp struct { + // MsID is the managed service ID to recover from. + MsID string `json:"msId"` + + // Time is the target timestamp in ISO 8601 format. + Time string `json:"time"` + + // Config is the backup store configuration. + Config BackupStoreConfig `json:"config"` + + // Secrets is the backup store secrets. + Secrets BackupStoreSecrets `json:"secrets"` +} + +// TakeBackupArgs contains arguments for taking a backup. +type TakeBackupArgs struct { + // ID is the backup ID. + ID string `json:"id"` + + // MsID is the managed service ID. + MsID string `json:"msId"` + + // Config is the backup store configuration. + Config BackupStoreConfig `json:"config"` + + // Secrets is the backup store secrets. + Secrets BackupStoreSecrets `json:"secrets"` +} diff --git a/model/common.go b/model/common.go index 81d97e9..4e565b9 100644 --- a/model/common.go +++ b/model/common.go @@ -35,65 +35,8 @@ type ServiceSecrets struct { SuperuserPassword string `json:"superuserPassword"` } -// BackupStoreConfig holds configuration for backup storage. -type BackupStoreConfig struct { - // EndpointURL is the S3-compatible endpoint URL. - EndpointURL string `json:"endpointUrl"` - - // DestinationPath is the path/bucket for backups. - DestinationPath string `json:"destinationPath"` - - // AccessKeyID is the access key ID for the backup store. - AccessKeyID string `json:"accessKeyId"` -} - -// BackupStoreSecrets holds sensitive data for backup storage. -type BackupStoreSecrets struct { - // SecretKey is the secret access key. - SecretKey string `json:"secretKey"` -} - -// RecoverFromBackup specifies recovery from a specific backup. -type RecoverFromBackup struct { - // ID is the backup ID to recover from. - ID string `json:"id"` - - // Config is the backup store configuration. - Config BackupStoreConfig `json:"config"` - - // Secrets is the backup store secrets. - Secrets BackupStoreSecrets `json:"secrets"` -} - -// RecoverFromTimestamp specifies point-in-time recovery. -type RecoverFromTimestamp struct { - // MsID is the managed service ID to recover from. - MsID string `json:"msId"` - - // Time is the target timestamp in ISO 8601 format. - Time string `json:"time"` - - // Config is the backup store configuration. - Config BackupStoreConfig `json:"config"` - - // Secrets is the backup store secrets. - Secrets BackupStoreSecrets `json:"secrets"` -} - // ServiceDetails contains common status details for a managed service. type ServiceDetails struct { // Ready indicates if the service is ready to accept connections. Ready bool `json:"ready"` } - -// TakeBackupArgs contains arguments for taking a backup. -type TakeBackupArgs struct { - // ID is the service ID. - ID string `json:"id"` - - // Config is the backup store configuration. - Config BackupStoreConfig `json:"config"` - - // Secrets is the backup store secrets. - Secrets BackupStoreSecrets `json:"secrets"` -} diff --git a/model/kubernetes.go b/model/kubernetes.go index 6eeb3b2..409ba7a 100644 --- a/model/kubernetes.go +++ b/model/kubernetes.go @@ -112,6 +112,10 @@ const ( // ServiceIDLabel is the label key for the managed service ID. ServiceIDLabel = LabelPrefix + "/id" + // BackupIDLabel is the label key for a backup ID. Applied to backup Jobs + // (alongside ServiceIDLabel) so a service's backups are discoverable. + BackupIDLabel = LabelPrefix + "/backup-id" + // ProviderLabel is the label key for the managed service provider type. ProviderLabel = LabelPrefix + "/provider" From af9cc8983a421427f1f7a24ce12d22c48173b360 Mon Sep 17 00:00:00 2001 From: utkuerol Date: Fri, 17 Jul 2026 20:17:44 +0200 Subject: [PATCH 2/4] add helpers for k8s jobs and also ServiceJob wrapper for specific provider operations --- README.md | 90 +++++++ client/jobrunner.go | 401 ++++++++++++++++++++++++++++++ client/jobrunner_test.go | 256 +++++++++++++++++++ provider/backupjob.go | 38 +++ provider/backupjob_test.go | 68 +++++ provider/base.go | 6 + provider/servicejob.go | 152 +++++++++++ provider/servicejob_test.go | 94 +++++++ provider/servicejob_usage_test.go | 149 +++++++++++ 9 files changed, 1254 insertions(+) create mode 100644 client/jobrunner.go create mode 100644 client/jobrunner_test.go create mode 100644 provider/backupjob.go create mode 100644 provider/backupjob_test.go create mode 100644 provider/servicejob.go create mode 100644 provider/servicejob_test.go create mode 100644 provider/servicejob_usage_test.go diff --git a/README.md b/README.md index 53fb0ec..3ed0f0b 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,96 @@ server.Run() `RegisterRoutes` mounts CRUD and backup endpoints under `/api/v1/{name}`. +## Detached jobs (backup & restore) + +Backups run **detached** from the provider pod as one-shot Kubernetes Jobs. `provider.Base` exposes a `Jobs` (`client.JobRunner`) helper for dispatching and polling them, and the backup lifecycle maps onto the Job lifecycle closely. The three backup methods collapse to near one-liners: + +```go +func (p *MyProvider) backupSpec(args model.TakeBackupArgs) client.JobSpec { + return provider.ServiceJobSpec(provider.ServiceJob{ + Operation: provider.JobOpBackup, // name: backup-, auto-labels BackupIDLabel + MsID: args.MsID, + Key: args.ID, + Image: p.backupImage, + Command: []string{"/backup"}, + // Env/Secrets are entirely yours — wire whatever your backup image reads. + Env: map[string]string{ + "BACKUP_STORE_ENDPOINT_URL": args.Config.EndpointURL, + "BACKUP_STORE_ACCESS_KEY_ID": args.Config.AccessKeyID, + }, + Secrets: map[string]string{"BACKUP_STORE_SECRET_KEY": args.Secrets.SecretKey}, + Labels: p.labels(args.MsID), + }) +} + +func (p *MyProvider) TakeBackup(ctx context.Context, args model.TakeBackupArgs) error { + return p.Jobs.Run(ctx, p.namespace(args.MsID), p.backupSpec(args)) +} + +func (p *MyProvider) GetBackupStatus(ctx context.Context, backupID string, retry model.TakeBackupArgs) (provider.BackupStatus, error) { + st, err := p.Jobs.State(ctx, p.namespace(retry.MsID), provider.BackupJobName(backupID)) + if err != nil { + return provider.BackupStatus{}, err + } + if st.Phase == client.JobFailed { + _ = p.Jobs.Replace(ctx, p.namespace(retry.MsID), p.backupSpec(retry)) + } + return provider.BackupStatusFromJob(st), nil +} + +func (p *MyProvider) DeleteBackup(ctx context.Context, args model.TakeBackupArgs) error { + return p.Jobs.Delete(ctx, p.namespace(args.MsID), provider.BackupJobName(args.ID)) +} +``` + +`ServiceJobSpec` owns the Job *conventions*: a deterministic name (`-`, so `Run` is idempotent and `State`/`Delete` resolve the same Job) and the identity labels (always `ServiceIDLabel`; backup operations also auto-stamp `BackupIDLabel` from the key). `Env` and `Secrets` pass through verbatim — each provider's image defines its own contract. Secrets are stored in the Job's owned Secret and injected via `secretKeyRef`, never appearing in the Job or Pod manifest. `Run` is idempotent on the Job name; `State` returns a phase (`pending`/`running`/`succeeded`/`failed`/`not_found`) and, on failure, folds the failed pod's logs into the reason; `Replace` is the retry path. Jobs default to a 6h deadline, a 3-retry backoff, and a 1h finished-TTL — override via `JobSpec` fields, or use `JobSpec.Customize` for anything not modelled. + +### Other detached operations + +Every detached operation is the same shape, so they all go through `provider.ServiceJob` — you supply the `Operation` prefix and execution details, and get the naming and identity-label conventions for free: + +| Operation | Job name | Auto-stamped label | +|-----------|----------|--------------------| +| `JobOpBackup` | `backup-` | `ServiceIDLabel`, `BackupIDLabel` | +| `JobOpDeleteBackup` | `delete-backup-` | `ServiceIDLabel`, `BackupIDLabel` | +| `JobOpRestore` | `restore-` | `ServiceIDLabel` | +| any provider-defined prefix | `-` | `ServiceIDLabel` | + +The operation→label policy lives in the lib, so a Job's identity labels always exist without the caller managing them; anything else you put in `Labels` rides along underneath. `ServiceIDLabel` is always stamped. The lib keeps only name helpers (`BackupJobName`, `DeleteBackupJobName`) and the status mappers (`BackupStatusFromJob` for the backup contract type, `OperationStatusFromJob` generically) — the three backup endpoints share these so they resolve the same Job. + +### Restore + +Restore is **not a standalone operation** in this contract — there is no restore endpoint. It happens **inside `Create`**: the create payload carries an optional `recoverFrom` (`model.RecoverFromBackup` or `model.RecoverFromTimestamp`, which the provider embeds in its own `CreateParams`), and when it is set the provider provisions the service *and* dispatches a detached Job to pull data from the backup store. Restore is keyed on the service ID (`restore-`) — one restore per service creation — and its progress is surfaced through the service's normal `GetStatus` (e.g. `details.ready = false` until it finishes), not a separate status endpoint. + +```go +func (p *MyProvider) Create(ctx context.Context, params MyCreateParams) error { + // ... provision the service (StatefulSet, Service, Secret, ...) ... + + if params.RecoverFrom != nil { + spec := provider.ServiceJobSpec(provider.ServiceJob{ + Operation: provider.JobOpRestore, // name: restore- + MsID: params.ID, + Key: params.ID, + Image: p.restoreImage, + Command: []string{"/restore"}, + // recoverFrom is provider-specific; wire whatever your restore image reads. + Env: map[string]string{"SOURCE_BACKUP_ID": params.RecoverFrom.Backup.ID}, + Secrets: map[string]string{"SOURCE_SECRET_KEY": params.RecoverFrom.Backup.Secrets.SecretKey}, + Labels: p.labels(params.ID), + }) + if err := p.Jobs.Run(ctx, p.namespace(params.ID), spec); err != nil { + return err + } + } + return nil +} + +// In GetStatus, fold the restore Job's progress into the service status: +// st, _ := p.Jobs.State(ctx, ns, provider.ServiceJobName(provider.JobOpRestore, id)) +// ready := st.Phase == client.JobSucceeded || st.Phase == client.JobNotFound +``` + + ## Configuration `config.Load()` reads these environment variables: diff --git a/client/jobrunner.go b/client/jobrunner.go new file mode 100644 index 0000000..0760eff --- /dev/null +++ b/client/jobrunner.go @@ -0,0 +1,401 @@ +// Copyright (c) Codesphere Inc. +// SPDX-License-Identifier: Apache-2.0 + +package client + +import ( + "context" + "errors" + "fmt" + "sort" + "time" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + + "github.com/codesphere-cloud/managed-services-lib/model" +) + +// Defaults applied to a JobSpec when the corresponding field is left zero. A +// backup job runs detached and must not run forever, so a bounded deadline is +// always set; a modest TTL keeps a finished Job around long enough for a status +// poll to observe the outcome before garbage collection. +const ( + DefaultJobTimeout = 6 * time.Hour + DefaultJobTTL = 1 * time.Hour + DefaultJobBackoffLimit = int32(3) + + // jobContainerName is the single container every dispatched Job runs. State + // reads logs from this container to explain failures. + jobContainerName = "job" + + // failureLogTailLines bounds how much of a failed pod's log is folded into + // the reported failure reason. + failureLogTailLines int64 = 20 +) + +// JobPhase is the coarse lifecycle state of a dispatched Job, derived from its +// status. It maps cleanly onto higher-level phase models (e.g. a backup phase). +type JobPhase string + +const ( + // JobNotFound means no Job of that name exists — never dispatched, or already + // garbage-collected after its TTL. + JobNotFound JobPhase = "not_found" + // JobPending means the Job exists but no pod is running yet. + JobPending JobPhase = "pending" + // JobRunning means at least one pod is active. + JobRunning JobPhase = "running" + // JobSucceeded means the Job completed successfully. + JobSucceeded JobPhase = "succeeded" + // JobFailed means the Job exhausted its retries or hit its deadline. + JobFailed JobPhase = "failed" +) + +// JobState is a point-in-time snapshot of a Job, shaped for a polling caller. +type JobState struct { + // Phase is the coarse lifecycle state. + Phase JobPhase + // StartedAt is when the Job first started a pod, if it has. + StartedAt *time.Time + // FinishedAt is when the Job reached a terminal phase, if it has. + FinishedAt *time.Time + // Reason explains a JobFailed phase: the failure condition message enriched, + // when available, with the tail of the failed pod's logs. Empty otherwise. + Reason string +} + +// JobSpec describes a one-shot Job to dispatch. Name is both the Job (and +// credentials Secret) name and the idempotency key: re-dispatching while a Job +// of the same name exists is a no-op rather than a duplicate. +type JobSpec struct { + // Name is the Job/Secret name and idempotency key. + Name string + // Image is the container image the Job runs. + Image string + // Command overrides the image entrypoint. + Command []string + // Args are the container arguments. + Args []string + // Env holds plain (non-secret) environment variables. + Env map[string]string + // Secrets holds secret environment variables. Each entry is stored in the + // Job's owned Secret and injected via a secretKeyRef, so values never appear + // in the Job or Pod manifest. The map keys are the env var names — there is + // no separate key list to keep in sync. + Secrets map[string]string + // Labels are applied to the Job and its Secret. Providers should set the + // standard service labels so the resources are discoverable and cleaned up + // with the service. + Labels map[string]string + + // Timeout bounds the Job's total run time (activeDeadlineSeconds). + // Zero means DefaultJobTimeout. + Timeout time.Duration + // TTL is how long a finished Job is kept before garbage collection + // (ttlSecondsAfterFinished). Zero means DefaultJobTTL. + TTL time.Duration + // BackoffLimit bounds pod retries. Nil means DefaultJobBackoffLimit. + BackoffLimit *int32 + // Resources sets the container resource requests/limits. + Resources corev1.ResourceRequirements + + // Customize, if set, is applied to the fully-built Job before it is created. + // It is the escape hatch for the rare field this struct does not model. + Customize func(*batchv1.Job) +} + +// JobRunner dispatches and inspects one-shot Kubernetes Jobs. +// It is a composed helper written against the KubernetesClient +// primitives, so it stays unit-testable against a mocked client and is reusable +// for any detached work (backups, restores, migrations). +type JobRunner struct { + kube KubernetesClient +} + +// NewJobRunner returns a JobRunner backed by the given client. +func NewJobRunner(kube KubernetesClient) JobRunner { + return JobRunner{kube: kube} +} + +// Run dispatches the Job described by spec into namespace. It is idempotent: if +// a Job of the same name already exists the call is a no-op. When spec.Secrets +// is non-empty a credentials Secret is created first (so the pod never starts +// without it) and then adopted by the Job for garbage collection. +func (r JobRunner) Run(ctx context.Context, namespace string, spec JobSpec) error { + hasSecret := len(spec.Secrets) > 0 + if hasSecret { + if _, err := r.kube.CreateSecret(ctx, namespace, buildJobSecret(spec, namespace)); err != nil && + !errors.Is(err, ErrResourceConflict) { + return fmt.Errorf("creating credentials secret for job %s: %w", spec.Name, err) + } + } + + job, err := buildJob(spec, namespace) + if err != nil { + return fmt.Errorf("building job %s: %w", spec.Name, err) + } + + created, err := r.kube.Create(ctx, model.JobAPI, namespace, job) + if err != nil { + if errors.Is(err, ErrResourceConflict) { + // Job already dispatched; its Secret already exists too. No-op. + return nil + } + return fmt.Errorf("creating job %s: %w", spec.Name, err) + } + + if hasSecret { + // Best-effort: adopt the Secret so it is garbage-collected with the Job. + // If this fails the Secret is still removed explicitly by Delete. + r.adoptSecret(ctx, namespace, spec.Name, created.GetUID()) + } + return nil +} + +// State returns a point-in-time snapshot of the named Job. A Job that does not +// exist is reported as JobNotFound with no error. A failed Job's Reason is +// enriched with the tail of the failed pod's logs when they can be read. +func (r JobRunner) State(ctx context.Context, namespace, name string) (JobState, error) { + obj, err := r.kube.Get(ctx, model.ResourceRef{APIResource: model.JobAPI, Namespace: namespace, Name: name}) + if err != nil { + if errors.Is(err, ErrResourceNotFound) { + return JobState{Phase: JobNotFound}, nil + } + return JobState{}, fmt.Errorf("getting job %s: %w", name, err) + } + + var job batchv1.Job + if err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.Object, &job); err != nil { + return JobState{}, fmt.Errorf("decoding job %s: %w", name, err) + } + + state := JobState{ + Phase: JobPending, + StartedAt: timePtr(job.Status.StartTime), + FinishedAt: timePtr(job.Status.CompletionTime), + } + switch { + case jobConditionTrue(job, batchv1.JobComplete): + state.Phase = JobSucceeded + case jobConditionTrue(job, batchv1.JobFailed): + state.Phase = JobFailed + state.Reason = r.failureReason(ctx, job) + if state.FinishedAt == nil { + state.FinishedAt = conditionTime(job, batchv1.JobFailed) + } + case job.Status.Active > 0: + state.Phase = JobRunning + } + return state, nil +} + +// Delete removes the named Job and its credentials Secret. Resources that no +// longer exist are not an error, so Delete is safe to call unconditionally. +func (r JobRunner) Delete(ctx context.Context, namespace, name string) error { + if err := r.kube.Delete(ctx, model.ResourceRef{APIResource: model.JobAPI, Namespace: namespace, Name: name}); err != nil && + !errors.Is(err, ErrResourceNotFound) { + return fmt.Errorf("deleting job %s: %w", name, err) + } + // The Secret is normally garbage-collected via its owner reference; delete it + // explicitly too in case adoption did not happen. + if err := r.kube.DeleteSecret(ctx, namespace, name); err != nil && + !errors.Is(err, ErrResourceNotFound) { + return fmt.Errorf("deleting credentials secret for job %s: %w", name, err) + } + return nil +} + +// Replace deletes any existing Job of the same name and dispatches a fresh one. +// It is the retry path: a failed Job must be removed before it can be re-run, +// because Run is a no-op while a Job of the same name exists. If the previous +// Job is still terminating the re-dispatch no-ops; call again on the next poll. +func (r JobRunner) Replace(ctx context.Context, namespace string, spec JobSpec) error { + if err := r.Delete(ctx, namespace, spec.Name); err != nil { + return err + } + return r.Run(ctx, namespace, spec) +} + +// adoptSecret patches the Job's owner reference onto its credentials Secret so +// the Secret is garbage-collected together with the Job. +// Best-effort: failures are tolerated because Delete removes the Secret explicitly as a fallback. +func (r JobRunner) adoptSecret(ctx context.Context, namespace, name string, jobUID types.UID) { + owner := []metav1.OwnerReference{{ + APIVersion: "batch/v1", + Kind: "Job", + Name: name, + UID: jobUID, + }} + _ = r.kube.Patch(ctx, model.ResourceRef{APIResource: model.SecretAPI, Namespace: namespace, Name: name}, + []model.K8sPatch{{Op: "add", Path: "/metadata/ownerReferences", Value: owner}}) +} + +// failureReason returns the Job's failure condition message, enriched with the +// tail of the failed pod's logs when they can be read. The condition alone +// ("BackoffLimitExceeded") rarely explains what went wrong; the logs do. +func (r JobRunner) failureReason(ctx context.Context, job batchv1.Job) string { + reason := conditionMessage(job, batchv1.JobFailed) + logs := r.failedPodLogs(ctx, job.Namespace, job.Name) + if logs == "" { + return reason + } + if reason == "" { + return logs + } + return reason + ": " + logs +} + +// failedPodLogs returns the tail of the logs of a pod belonging to the Job. +// Best-effort: any error yields an empty string. +func (r JobRunner) failedPodLogs(ctx context.Context, namespace, jobName string) string { + pods, err := r.kube.List(ctx, model.ListOptions{ + APIResource: model.PodAPI, + Namespace: namespace, + LabelSelector: "job-name=" + jobName, + }) + if err != nil || pods == nil || len(pods.Items) == 0 { + return "" + } + // The most recently created pod is the one that exhausted the retries. + podName := pods.Items[len(pods.Items)-1].GetName() + logs, err := r.kube.GetPodLogs(ctx, namespace, podName, jobContainerName, failureLogTailLines) + if err != nil { + return "" + } + return logs +} + +func buildJob(spec JobSpec, namespace string) (*unstructured.Unstructured, error) { + ttl := int32(durationSeconds(spec.TTL, DefaultJobTTL)) + deadline := durationSeconds(spec.Timeout, DefaultJobTimeout) + backoff := DefaultJobBackoffLimit + if spec.BackoffLimit != nil { + backoff = *spec.BackoffLimit + } + + job := &batchv1.Job{ + TypeMeta: metav1.TypeMeta{APIVersion: "batch/v1", Kind: "Job"}, + ObjectMeta: metav1.ObjectMeta{Name: spec.Name, Namespace: namespace, Labels: spec.Labels}, + Spec: batchv1.JobSpec{ + TTLSecondsAfterFinished: &ttl, + ActiveDeadlineSeconds: &deadline, + BackoffLimit: &backoff, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: spec.Labels}, + Spec: corev1.PodSpec{ + RestartPolicy: corev1.RestartPolicyOnFailure, + Containers: []corev1.Container{{ + Name: jobContainerName, + Image: spec.Image, + ImagePullPolicy: corev1.PullIfNotPresent, + Command: spec.Command, + Args: spec.Args, + Env: buildEnv(spec), + Resources: spec.Resources, + }}, + }, + }, + }, + } + if spec.Customize != nil { + spec.Customize(job) + } + + m, err := runtime.DefaultUnstructuredConverter.ToUnstructured(job) + if err != nil { + return nil, fmt.Errorf("converting job to unstructured: %w", err) + } + return &unstructured.Unstructured{Object: m}, nil +} + +func buildJobSecret(spec JobSpec, namespace string) *corev1.Secret { + return &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: spec.Name, Namespace: namespace, Labels: spec.Labels}, + Type: corev1.SecretTypeOpaque, + StringData: spec.Secrets, + } +} + +// buildEnv builds the container env: plain vars from Env, then secret vars from +// Secrets routed through a secretKeyRef against the Job's own Secret. Both are +// emitted in sorted key order so the produced Job is deterministic. +func buildEnv(spec JobSpec) []corev1.EnvVar { + out := make([]corev1.EnvVar, 0, len(spec.Env)+len(spec.Secrets)) + for _, k := range sortedKeys(spec.Env) { + out = append(out, corev1.EnvVar{Name: k, Value: spec.Env[k]}) + } + for _, k := range sortedKeys(spec.Secrets) { + out = append(out, corev1.EnvVar{ + Name: k, + ValueFrom: &corev1.EnvVarSource{ + SecretKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: spec.Name}, + Key: k, + }, + }, + }) + } + return out +} + +func sortedKeys(m map[string]string) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +func durationSeconds(d, fallback time.Duration) int64 { + if d <= 0 { + d = fallback + } + return int64(d.Seconds()) +} + +func jobConditionTrue(job batchv1.Job, t batchv1.JobConditionType) bool { + for _, c := range job.Status.Conditions { + if c.Type == t { + return c.Status == corev1.ConditionTrue + } + } + return false +} + +func conditionMessage(job batchv1.Job, t batchv1.JobConditionType) string { + for _, c := range job.Status.Conditions { + if c.Type != t { + continue + } + if c.Message != "" { + return c.Message + } + if c.Reason != "" { + return c.Reason + } + } + return "unknown reason" +} + +func conditionTime(job batchv1.Job, t batchv1.JobConditionType) *time.Time { + for _, c := range job.Status.Conditions { + if c.Type == t && !c.LastTransitionTime.IsZero() { + return timePtr(&c.LastTransitionTime) + } + } + return nil +} + +func timePtr(t *metav1.Time) *time.Time { + if t == nil || t.IsZero() { + return nil + } + out := t.Time + return &out +} diff --git a/client/jobrunner_test.go b/client/jobrunner_test.go new file mode 100644 index 0000000..0f74d7f --- /dev/null +++ b/client/jobrunner_test.go @@ -0,0 +1,256 @@ +// Copyright (c) Codesphere Inc. +// SPDX-License-Identifier: Apache-2.0 + +package client_test + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/stretchr/testify/mock" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + + "github.com/codesphere-cloud/managed-services-lib/client" + "github.com/codesphere-cloud/managed-services-lib/client/mocks" + "github.com/codesphere-cloud/managed-services-lib/model" +) + +func toUnstructured(obj interface{}) *unstructured.Unstructured { + m, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj) + Expect(err).NotTo(HaveOccurred()) + return &unstructured.Unstructured{Object: m} +} + +func jobFrom(u *unstructured.Unstructured) batchv1.Job { + var job batchv1.Job + Expect(runtime.DefaultUnstructuredConverter.FromUnstructured(u.Object, &job)).To(Succeed()) + return job +} + +// jobWithStatus builds an unstructured Job carrying the given status, as Get would return. +func jobWithStatus(name, namespace string, status batchv1.JobStatus) *unstructured.Unstructured { + return toUnstructured(&batchv1.Job{ + TypeMeta: metav1.TypeMeta{APIVersion: "batch/v1", Kind: "Job"}, + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}, + Status: status, + }) +} + +var _ = Describe("JobRunner", func() { + var ( + ctx context.Context + kube *mocks.MockKubernetesClient + runner client.JobRunner + spec client.JobSpec + ) + + BeforeEach(func() { + ctx = context.Background() + kube = new(mocks.MockKubernetesClient) + runner = client.NewJobRunner(kube) + spec = client.JobSpec{ + Name: "s3-backup-x", + Image: "job-image:1", + Command: []string{"/app/backup-job"}, + Labels: map[string]string{"backup-id": "x"}, + Env: map[string]string{"BACKUP_ID": "x"}, + Secrets: map[string]string{"STORE_SECRET_KEY": "sk"}, + } + }) + + AfterEach(func() { kube.AssertExpectations(GinkgoT()) }) + + Describe("Run", func() { + It("creates the secret, then the job with the right container, env and limits", func() { + var createdJob *unstructured.Unstructured + var createdSecret *corev1.Secret + created := toUnstructured(&batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: spec.Name}}) + created.SetUID("job-uid-123") + + kube.On("CreateSecret", mock.Anything, "jobs-ns", mock.Anything). + Run(func(a mock.Arguments) { createdSecret = a.Get(2).(*corev1.Secret) }). + Return(&corev1.Secret{}, nil) + kube.On("Create", mock.Anything, model.JobAPI, "jobs-ns", mock.Anything). + Run(func(a mock.Arguments) { createdJob = a.Get(3).(*unstructured.Unstructured) }). + Return(created, nil) + kube.On("Patch", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + Expect(runner.Run(ctx, "jobs-ns", spec)).To(Succeed()) + + job := jobFrom(createdJob) + Expect(job.Namespace).To(Equal("jobs-ns")) + Expect(*job.Spec.BackoffLimit).To(Equal(int32(3))) + Expect(*job.Spec.TTLSecondsAfterFinished).To(Equal(int32(3600))) + Expect(*job.Spec.ActiveDeadlineSeconds).To(Equal(int64(6 * 60 * 60))) + + c := job.Spec.Template.Spec.Containers[0] + Expect(c.Image).To(Equal("job-image:1")) + Expect(c.Command).To(Equal([]string{"/app/backup-job"})) + Expect(job.Spec.Template.Spec.RestartPolicy).To(Equal(corev1.RestartPolicyOnFailure)) + + env := map[string]corev1.EnvVar{} + for _, e := range c.Env { + env[e.Name] = e + } + Expect(env["BACKUP_ID"].Value).To(Equal("x")) + Expect(env["STORE_SECRET_KEY"].Value).To(BeEmpty()) + Expect(env["STORE_SECRET_KEY"].ValueFrom.SecretKeyRef.Name).To(Equal("s3-backup-x")) + Expect(env["STORE_SECRET_KEY"].ValueFrom.SecretKeyRef.Key).To(Equal("STORE_SECRET_KEY")) + + Expect(createdSecret.Name).To(Equal("s3-backup-x")) + Expect(createdSecret.StringData).To(HaveKeyWithValue("STORE_SECRET_KEY", "sk")) + }) + + It("adopts the secret under the created job for garbage collection", func() { + var ref model.ResourceRef + var patches []model.K8sPatch + created := toUnstructured(&batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: spec.Name}}) + created.SetUID("job-uid-123") + + kube.On("CreateSecret", mock.Anything, mock.Anything, mock.Anything).Return(&corev1.Secret{}, nil) + kube.On("Create", mock.Anything, model.JobAPI, mock.Anything, mock.Anything).Return(created, nil) + kube.On("Patch", mock.Anything, mock.Anything, mock.Anything). + Run(func(a mock.Arguments) { + ref = a.Get(1).(model.ResourceRef) + patches = a.Get(2).([]model.K8sPatch) + }).Return(nil) + + Expect(runner.Run(ctx, "jobs-ns", spec)).To(Succeed()) + + Expect(ref.APIResource).To(Equal(model.SecretAPI)) + Expect(ref.Name).To(Equal("s3-backup-x")) + Expect(patches).To(HaveLen(1)) + owners := patches[0].Value.([]metav1.OwnerReference) + Expect(owners[0].Kind).To(Equal("Job")) + Expect(string(owners[0].UID)).To(Equal("job-uid-123")) + }) + + It("is a no-op if the job already exists", func() { + kube.On("CreateSecret", mock.Anything, mock.Anything, mock.Anything).Return(&corev1.Secret{}, nil) + kube.On("Create", mock.Anything, model.JobAPI, mock.Anything, mock.Anything). + Return(nil, client.ErrResourceConflict) + // No Patch: adoption is skipped when the job was not freshly created. + + Expect(runner.Run(ctx, "jobs-ns", spec)).To(Succeed()) + kube.AssertNotCalled(GinkgoT(), "Patch", mock.Anything, mock.Anything, mock.Anything) + }) + + It("creates no secret when the spec has none", func() { + spec.Secrets = nil + created := toUnstructured(&batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: spec.Name}}) + kube.On("Create", mock.Anything, model.JobAPI, mock.Anything, mock.Anything).Return(created, nil) + + Expect(runner.Run(ctx, "jobs-ns", spec)).To(Succeed()) + kube.AssertNotCalled(GinkgoT(), "CreateSecret", mock.Anything, mock.Anything, mock.Anything) + kube.AssertNotCalled(GinkgoT(), "Patch", mock.Anything, mock.Anything, mock.Anything) + }) + }) + + Describe("State", func() { + It("reports a succeeded job", func() { + done := metav1.Now() + kube.On("Get", mock.Anything, mock.Anything).Return(jobWithStatus("s3-backup-x", "jobs-ns", batchv1.JobStatus{ + CompletionTime: &done, + Conditions: []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue}}, + }), nil) + + st, err := runner.State(ctx, "jobs-ns", "s3-backup-x") + Expect(err).NotTo(HaveOccurred()) + Expect(st.Phase).To(Equal(client.JobSucceeded)) + Expect(st.FinishedAt).NotTo(BeNil()) + }) + + It("reports a running job", func() { + kube.On("Get", mock.Anything, mock.Anything).Return(jobWithStatus("s3-backup-x", "jobs-ns", batchv1.JobStatus{ + Active: 1, + }), nil) + + st, err := runner.State(ctx, "jobs-ns", "s3-backup-x") + Expect(err).NotTo(HaveOccurred()) + Expect(st.Phase).To(Equal(client.JobRunning)) + }) + + It("reports a freshly created job as pending", func() { + kube.On("Get", mock.Anything, mock.Anything).Return(jobWithStatus("s3-backup-x", "jobs-ns", batchv1.JobStatus{}), nil) + + st, err := runner.State(ctx, "jobs-ns", "s3-backup-x") + Expect(err).NotTo(HaveOccurred()) + Expect(st.Phase).To(Equal(client.JobPending)) + }) + + It("reports a failed job, enriching the reason with pod logs", func() { + kube.On("Get", mock.Anything, mock.Anything).Return(jobWithStatus("s3-backup-x", "jobs-ns", batchv1.JobStatus{ + Conditions: []batchv1.JobCondition{{ + Type: batchv1.JobFailed, Status: corev1.ConditionTrue, Message: "BackoffLimitExceeded", + }}, + }), nil) + pod := toUnstructured(&corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "s3-backup-x-abc"}}) + kube.On("List", mock.Anything, mock.MatchedBy(func(o model.ListOptions) bool { + return o.APIResource == model.PodAPI && o.LabelSelector == "job-name=s3-backup-x" + })).Return(&unstructured.UnstructuredList{Items: []unstructured.Unstructured{*pod}}, nil) + kube.On("GetPodLogs", mock.Anything, "jobs-ns", "s3-backup-x-abc", "job", int64(20)). + Return("connection refused", nil) + + st, err := runner.State(ctx, "jobs-ns", "s3-backup-x") + Expect(err).NotTo(HaveOccurred()) + Expect(st.Phase).To(Equal(client.JobFailed)) + Expect(st.Reason).To(Equal("BackoffLimitExceeded: connection refused")) + }) + + It("reports a missing job as not found", func() { + kube.On("Get", mock.Anything, mock.Anything).Return(nil, client.ErrResourceNotFound) + + st, err := runner.State(ctx, "jobs-ns", "s3-backup-x") + Expect(err).NotTo(HaveOccurred()) + Expect(st.Phase).To(Equal(client.JobNotFound)) + }) + }) + + Describe("Delete", func() { + It("deletes the job and its secret", func() { + var deletedJob bool + var deletedSecret string + kube.On("Delete", mock.Anything, mock.MatchedBy(func(r model.ResourceRef) bool { + return r.APIResource == model.JobAPI + })).Run(func(mock.Arguments) { deletedJob = true }).Return(nil) + kube.On("DeleteSecret", mock.Anything, "jobs-ns", mock.Anything). + Run(func(a mock.Arguments) { deletedSecret = a.Get(2).(string) }).Return(nil) + + Expect(runner.Delete(ctx, "jobs-ns", "s3-backup-x")).To(Succeed()) + Expect(deletedJob).To(BeTrue()) + Expect(deletedSecret).To(Equal("s3-backup-x")) + }) + + It("tolerates a job and secret that do not exist", func() { + kube.On("Delete", mock.Anything, mock.Anything).Return(client.ErrResourceNotFound) + kube.On("DeleteSecret", mock.Anything, mock.Anything, mock.Anything).Return(client.ErrResourceNotFound) + + Expect(runner.Delete(ctx, "jobs-ns", "s3-backup-x")).To(Succeed()) + }) + }) + + Describe("Replace", func() { + It("deletes the existing job before dispatching a fresh one", func() { + calls := []string{} + kube.On("Delete", mock.Anything, mock.Anything). + Run(func(mock.Arguments) { calls = append(calls, "delete-job") }).Return(nil) + kube.On("DeleteSecret", mock.Anything, mock.Anything, mock.Anything). + Run(func(mock.Arguments) { calls = append(calls, "delete-secret") }).Return(nil) + kube.On("CreateSecret", mock.Anything, mock.Anything, mock.Anything). + Run(func(mock.Arguments) { calls = append(calls, "create-secret") }).Return(&corev1.Secret{}, nil) + created := toUnstructured(&batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: spec.Name}}) + kube.On("Create", mock.Anything, model.JobAPI, mock.Anything, mock.Anything). + Run(func(mock.Arguments) { calls = append(calls, "create-job") }).Return(created, nil) + kube.On("Patch", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + Expect(runner.Replace(ctx, "jobs-ns", spec)).To(Succeed()) + Expect(calls).To(Equal([]string{"delete-job", "delete-secret", "create-secret", "create-job"})) + }) + }) +}) diff --git a/provider/backupjob.go b/provider/backupjob.go new file mode 100644 index 0000000..8ee34e6 --- /dev/null +++ b/provider/backupjob.go @@ -0,0 +1,38 @@ +// Copyright (c) Codesphere Inc. +// SPDX-License-Identifier: Apache-2.0 + +package provider + +import ( + "github.com/codesphere-cloud/managed-services-lib/client" +) + +// Backup is a framework contract operation (TakeBackup / GetBackupStatus / +// DeleteBackup), so the lib owns the naming and status conventions its three +// endpoints must agree on. Build the Job specs with ServiceJob using the +// JobOpBackup / JobOpDeleteBackup prefixes — both auto-stamp the backup identity +// labels; these helpers only resolve names and status. + +// BackupJobName is the name of the Job that takes a backup. +func BackupJobName(backupID string) string { + return ServiceJobName(JobOpBackup, backupID) +} + +// DeleteBackupJobName is the name of the Job that deletes a backup. +func DeleteBackupJobName(backupID string) string { + return ServiceJobName(JobOpDeleteBackup, backupID) +} + +// BackupStatusFromJob maps a Job snapshot onto the backup status contract. A Job +// that no longer exists is reported as pending — a provider that needs to +// distinguish "never taken" from "completed and garbage-collected" should check +// the JobState phase directly before calling this. +func BackupStatusFromJob(s client.JobState) BackupStatus { + op := OperationStatusFromJob(s) + return BackupStatus{ + Phase: BackupPhase(op.Phase), + StartedAt: op.StartedAt, + CompletedAt: op.CompletedAt, + Error: op.Error, + } +} diff --git a/provider/backupjob_test.go b/provider/backupjob_test.go new file mode 100644 index 0000000..c698450 --- /dev/null +++ b/provider/backupjob_test.go @@ -0,0 +1,68 @@ +// Copyright (c) Codesphere Inc. +// SPDX-License-Identifier: Apache-2.0 + +package provider_test + +import ( + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/codesphere-cloud/managed-services-lib/client" + "github.com/codesphere-cloud/managed-services-lib/model" + "github.com/codesphere-cloud/managed-services-lib/provider" +) + +var _ = Describe("Backup job helpers", func() { + Describe("Job names", func() { + It("names the take- and delete-backup jobs by backup ID, distinctly", func() { + Expect(provider.BackupJobName("bkp-7")).To(Equal("backup-bkp-7")) + Expect(provider.DeleteBackupJobName("bkp-7")).To(Equal("delete-backup-bkp-7")) + Expect(provider.BackupJobName("bkp-7")).NotTo(Equal(provider.DeleteBackupJobName("bkp-7"))) + }) + + It("agrees with the name a ServiceJob of that operation produces", func() { + spec := provider.ServiceJobSpec(provider.ServiceJob{ + Operation: provider.JobOpBackup, MsID: "svc-42", Key: "bkp-7", + }) + Expect(spec.Name).To(Equal(provider.BackupJobName("bkp-7"))) + // The backup operation auto-stamps the backup identity label. + Expect(spec.Labels).To(HaveKeyWithValue(model.ServiceIDLabel, "svc-42")) + Expect(spec.Labels).To(HaveKeyWithValue(model.BackupIDLabel, "bkp-7")) + }) + }) + + Describe("BackupStatusFromJob", func() { + started := time.Date(2026, 7, 17, 10, 0, 0, 0, time.UTC) + finished := time.Date(2026, 7, 17, 10, 5, 0, 0, time.UTC) + + It("maps a running job", func() { + s := provider.BackupStatusFromJob(client.JobState{Phase: client.JobRunning, StartedAt: &started}) + Expect(s.Phase).To(Equal(provider.BackupPhaseRunning)) + Expect(s.StartedAt).To(Equal("2026-07-17T10:00:00Z")) + Expect(s.Error).To(BeEmpty()) + }) + + It("maps a succeeded job with timestamps", func() { + s := provider.BackupStatusFromJob(client.JobState{ + Phase: client.JobSucceeded, StartedAt: &started, FinishedAt: &finished, + }) + Expect(s.Phase).To(Equal(provider.BackupPhaseCompleted)) + Expect(s.CompletedAt).To(Equal("2026-07-17T10:05:00Z")) + }) + + It("maps a failed job, carrying the reason into Error", func() { + s := provider.BackupStatusFromJob(client.JobState{Phase: client.JobFailed, Reason: "boom"}) + Expect(s.Phase).To(Equal(provider.BackupPhaseFailed)) + Expect(s.Error).To(Equal("boom")) + }) + + It("maps pending and not-found to pending", func() { + Expect(provider.BackupStatusFromJob(client.JobState{Phase: client.JobPending}).Phase). + To(Equal(provider.BackupPhasePending)) + Expect(provider.BackupStatusFromJob(client.JobState{Phase: client.JobNotFound}).Phase). + To(Equal(provider.BackupPhasePending)) + }) + }) +}) diff --git a/provider/base.go b/provider/base.go index cd9ca27..345bdeb 100644 --- a/provider/base.go +++ b/provider/base.go @@ -50,6 +50,11 @@ type KubernetesClient = client.KubernetesClient type Base struct { K8sClient KubernetesClient Logger *slog.Logger + + // Jobs dispatches and inspects detached one-shot Jobs (e.g. backup/restore + // work that must run outside the provider pod). Build specs with ServiceJob; + // see BackupStatusFromJob for the backup status contract. + Jobs client.JobRunner } // NewBase creates a new Base provider. @@ -57,6 +62,7 @@ func NewBase(k8sClient KubernetesClient, logger *slog.Logger) *Base { return &Base{ K8sClient: k8sClient, Logger: logger, + Jobs: client.NewJobRunner(k8sClient), } } diff --git a/provider/servicejob.go b/provider/servicejob.go new file mode 100644 index 0000000..199c2ba --- /dev/null +++ b/provider/servicejob.go @@ -0,0 +1,152 @@ +// Copyright (c) Codesphere Inc. +// SPDX-License-Identifier: Apache-2.0 + +package provider + +import ( + "maps" + "time" + + "github.com/codesphere-cloud/managed-services-lib/client" + "github.com/codesphere-cloud/managed-services-lib/model" +) + +// Operation prefixes for detached service Jobs. Each becomes the Job name prefix +// and identifies the kind of work. Providers running other detached work can use +// their own prefix with ServiceJob directly. +const ( + JobOpBackup = "backup" + JobOpDeleteBackup = "delete-backup" + JobOpRestore = "restore" +) + +// operationKeyLabels maps an operation to the identity label its Key is stamped +// under, so every Job of that kind carries a consistent, discoverable label +// without the caller having to remember it. Operations absent from this map +// (restore, provider-defined kinds) carry only the ServiceIDLabel. +var operationKeyLabels = map[string]string{ + JobOpBackup: model.BackupIDLabel, + JobOpDeleteBackup: model.BackupIDLabel, +} + +// ServiceJob describes a detached one-shot operation run on behalf of a managed +// service — a backup, backup deletion, restore-on-create and so on. +// It is the single entry point for dispatching such work: it applies the Job +// conventions (naming and identity labels) so every provider's operations are +// named and labelled consistently. Execution details (Image/Command/Env/Secrets) +// pass through verbatim: each provider's operation image defines its own contract. +type ServiceJob struct { + // Operation is the job kind; it becomes the Job name prefix (one of the JobOp + // constants, or a provider-defined prefix). Known operations also fix which + // identity label the Key is stamped under (see operationKeyLabels). + Operation string + // MsID is the service the job acts on. It is always stamped as ServiceIDLabel + // so the Job is discoverable and cleaned up with the service. + MsID string + // Key uniquely identifies the job within its Operation, becoming the Job name + // suffix and idempotency key. Use the backup ID for per-backup operations, or + // the MsID for whole-service operations (e.g. restore). + Key string + // Image is the container image the Job runs. + Image string + // Command overrides the image entrypoint. + Command []string + // Env holds plain (non-secret) environment variables. + Env map[string]string + // Secrets holds secret environment variables. Each is stored in the Job's + // owned Secret and injected via a secretKeyRef, so values never appear in the + // Job or Pod manifest. + Secrets map[string]string + // Labels are merged under the identity labels, which the caller cannot + // override. + Labels map[string]string +} + +// ServiceJobName is the Job naming convention: "-". It is stable +// per operation and key so Run is idempotent and State/Delete resolve the same +// Job. Keep keys short: the result must stay within the 63-char DNS-1123 limit. +func ServiceJobName(operation, key string) string { + return operation + "-" + key +} + +// ServiceJobSpec applies the naming and identity-label conventions to build a +// client.JobSpec. It always stamps ServiceIDLabel from MsID, and — for +// operations with a registered key label (see operationKeyLabels) — the Key +// under that label. Caller Labels ride along underneath and cannot override the +// identity labels. +func ServiceJobSpec(j ServiceJob) client.JobSpec { + labels := map[string]string{} + maps.Copy(labels, j.Labels) + // Identity labels are authoritative — set them last so caller labels cannot + // override them. + labels[model.ServiceIDLabel] = j.MsID + if keyLabel := operationKeyLabels[j.Operation]; keyLabel != "" { + labels[keyLabel] = j.Key + } + return client.JobSpec{ + Name: ServiceJobName(j.Operation, j.Key), + Image: j.Image, + Command: j.Command, + Env: j.Env, + Secrets: j.Secrets, + Labels: labels, + } +} + +// OperationStatus is the phase snapshot shared by every detached operation. It +// is derived from a client.JobState. The backup REST contract has its own +// BackupStatus (built from this); other operations can surface it directly. +type OperationStatus struct { + // Phase is the current phase of the operation. + Phase OperationPhase `json:"phase"` + + // StartedAt is when the operation started. + StartedAt string `json:"startedAt,omitempty"` + + // CompletedAt is when the operation completed. + CompletedAt string `json:"completedAt,omitempty"` + + // Error contains any error message. + Error string `json:"error,omitempty"` +} + +// OperationPhase represents the phase of a detached operation. Its values match +// the backup REST contract's phases so BackupStatus can be built from it. +type OperationPhase string + +// Operation phase constants. +const ( + OperationPhasePending OperationPhase = "pending" + OperationPhaseRunning OperationPhase = "running" + OperationPhaseCompleted OperationPhase = "completed" + OperationPhaseFailed OperationPhase = "failed" +) + +// OperationStatusFromJob maps a Job snapshot onto an OperationStatus. A Job that +// no longer exists is reported as pending — a caller that needs to distinguish +// "never run" from "completed and garbage-collected" should check the JobState +// phase directly before calling this. +func OperationStatusFromJob(s client.JobState) OperationStatus { + status := OperationStatus{ + Phase: OperationPhasePending, + StartedAt: formatTime(s.StartedAt), + CompletedAt: formatTime(s.FinishedAt), + } + switch s.Phase { + case client.JobRunning: + status.Phase = OperationPhaseRunning + case client.JobSucceeded: + status.Phase = OperationPhaseCompleted + case client.JobFailed: + status.Phase = OperationPhaseFailed + status.Error = s.Reason + } + return status +} + +func formatTime(t *time.Time) string { + if t == nil { + return "" + } + return t.UTC().Format(time.RFC3339) +} diff --git a/provider/servicejob_test.go b/provider/servicejob_test.go new file mode 100644 index 0000000..a0822ee --- /dev/null +++ b/provider/servicejob_test.go @@ -0,0 +1,94 @@ +// Copyright (c) Codesphere Inc. +// SPDX-License-Identifier: Apache-2.0 + +package provider_test + +import ( + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/codesphere-cloud/managed-services-lib/client" + "github.com/codesphere-cloud/managed-services-lib/model" + "github.com/codesphere-cloud/managed-services-lib/provider" +) + +var _ = Describe("Service job core", func() { + Describe("ServiceJobSpec", func() { + It("names the job - and always stamps the service label", func() { + spec := provider.ServiceJobSpec(provider.ServiceJob{ + Operation: "something", + MsID: "svc-42", + Key: "svc-42", + Image: "something:1", + }) + + Expect(spec.Name).To(Equal("something-svc-42")) + Expect(spec.Name).To(Equal(provider.ServiceJobName("something", "svc-42"))) + Expect(spec.Labels).To(HaveKeyWithValue(model.ServiceIDLabel, "svc-42")) + }) + + It("auto-stamps the key label for operations that have one", func() { + // Backup-family operations stamp the backup identity label from Key... + for _, op := range []string{provider.JobOpBackup, provider.JobOpDeleteBackup} { + spec := provider.ServiceJobSpec(provider.ServiceJob{Operation: op, MsID: "svc-42", Key: "bkp-7"}) + Expect(spec.Labels).To(HaveKeyWithValue(model.BackupIDLabel, "bkp-7")) + } + + // ...operations without a registered label (restore, custom) do not. + for _, op := range []string{provider.JobOpRestore, "custom"} { + spec := provider.ServiceJobSpec(provider.ServiceJob{Operation: op, MsID: "svc-42", Key: "svc-42"}) + Expect(spec.Labels).NotTo(HaveKey(model.BackupIDLabel)) + } + }) + + It("lets caller labels ride along but never override identity labels", func() { + spec := provider.ServiceJobSpec(provider.ServiceJob{ + Operation: provider.JobOpBackup, MsID: "svc-42", Key: "bkp-7", + Labels: map[string]string{ + "custom": "yes", + model.ServiceIDLabel: "hijack", + model.BackupIDLabel: "hijack", + }, + }) + + Expect(spec.Labels).To(HaveKeyWithValue("custom", "yes")) + Expect(spec.Labels).To(HaveKeyWithValue(model.ServiceIDLabel, "svc-42")) + Expect(spec.Labels).To(HaveKeyWithValue(model.BackupIDLabel, "bkp-7")) + }) + }) + + Describe("OperationStatusFromJob", func() { + started := time.Date(2026, 7, 17, 10, 0, 0, 0, time.UTC) + finished := time.Date(2026, 7, 17, 10, 5, 0, 0, time.UTC) + + It("maps a running job", func() { + s := provider.OperationStatusFromJob(client.JobState{Phase: client.JobRunning, StartedAt: &started}) + Expect(s.Phase).To(Equal(provider.OperationPhaseRunning)) + Expect(s.StartedAt).To(Equal("2026-07-17T10:00:00Z")) + Expect(s.Error).To(BeEmpty()) + }) + + It("maps a succeeded job with timestamps", func() { + s := provider.OperationStatusFromJob(client.JobState{ + Phase: client.JobSucceeded, StartedAt: &started, FinishedAt: &finished, + }) + Expect(s.Phase).To(Equal(provider.OperationPhaseCompleted)) + Expect(s.CompletedAt).To(Equal("2026-07-17T10:05:00Z")) + }) + + It("maps a failed job, carrying the reason into Error", func() { + s := provider.OperationStatusFromJob(client.JobState{Phase: client.JobFailed, Reason: "boom"}) + Expect(s.Phase).To(Equal(provider.OperationPhaseFailed)) + Expect(s.Error).To(Equal("boom")) + }) + + It("maps pending and not-found to pending", func() { + Expect(provider.OperationStatusFromJob(client.JobState{Phase: client.JobPending}).Phase). + To(Equal(provider.OperationPhasePending)) + Expect(provider.OperationStatusFromJob(client.JobState{Phase: client.JobNotFound}).Phase). + To(Equal(provider.OperationPhasePending)) + }) + }) +}) diff --git a/provider/servicejob_usage_test.go b/provider/servicejob_usage_test.go new file mode 100644 index 0000000..1964b05 --- /dev/null +++ b/provider/servicejob_usage_test.go @@ -0,0 +1,149 @@ +// Copyright (c) Codesphere Inc. +// SPDX-License-Identifier: Apache-2.0 + +package provider_test + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/stretchr/testify/mock" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + + "github.com/codesphere-cloud/managed-services-lib/client" + "github.com/codesphere-cloud/managed-services-lib/client/mocks" + "github.com/codesphere-cloud/managed-services-lib/model" + "github.com/codesphere-cloud/managed-services-lib/provider" +) + +func runningJob(name, namespace string) *unstructured.Unstructured { + start := metav1.Now() + m, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&batchv1.Job{ + TypeMeta: metav1.TypeMeta{APIVersion: "batch/v1", Kind: "Job"}, + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}, + Status: batchv1.JobStatus{Active: 1, StartTime: &start}, + }) + Expect(err).NotTo(HaveOccurred()) + return &unstructured.Unstructured{Object: m} +} + +// This suite is a worked example of how a provider dispatches detached work: +// build the spec with ServiceJob (which applies the naming + label conventions), +// then Run/State/Delete it through Base.Jobs (a client.JobRunner). It covers the +// three operations a provider actually runs as Jobs — take backup, delete +// backup, and restore-on-create — plus polling and cleanup. +var _ = Describe("Dispatching operations with ServiceJob + JobRunner", func() { + const namespace = "rg-7" // provider.NamespaceForTeam(7) + + var ( + ctx context.Context + kube *mocks.MockKubernetesClient + runner client.JobRunner + created *unstructured.Unstructured + ) + + BeforeEach(func() { + ctx = context.Background() + kube = new(mocks.MockKubernetesClient) + runner = client.NewJobRunner(kube) + created = nil + }) + + // stubDispatch stubs the create path (owned secret -> job -> adopt) and + // captures the Job object that reaches the cluster. + stubDispatch := func() { + kube.On("CreateSecret", mock.Anything, mock.Anything, mock.Anything).Return(&corev1.Secret{}, nil).Maybe() + kube.On("Patch", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() + withUID := &unstructured.Unstructured{Object: map[string]any{}} + withUID.SetUID("job-uid") // returned to Run for secret adoption + kube.On("Create", mock.Anything, model.JobAPI, mock.Anything, mock.Anything). + Run(func(a mock.Arguments) { created = a.Get(3).(*unstructured.Unstructured) }). + Return(withUID, nil) + } + + It("TakeBackup: dispatches backup-, labelled by service and backup", func() { + stubDispatch() + + spec := provider.ServiceJobSpec(provider.ServiceJob{ + Operation: provider.JobOpBackup, + MsID: "svc-42", + Key: "bkp-7", // the backup ID + Image: "backup-image:1", + Command: []string{"/backup"}, + Env: map[string]string{"BACKUP_STORE_ENDPOINT_URL": "https://s3.example.com"}, + Secrets: map[string]string{"BACKUP_STORE_SECRET_KEY": "shh"}, + Labels: map[string]string{model.TeamIDLabel: "7"}, + }) + Expect(runner.Run(ctx, namespace, spec)).To(Succeed()) + + Expect(created.GetName()).To(Equal("backup-bkp-7")) + Expect(created.GetLabels()).To(HaveKeyWithValue(model.ServiceIDLabel, "svc-42")) + Expect(created.GetLabels()).To(HaveKeyWithValue(model.BackupIDLabel, "bkp-7")) + Expect(created.GetLabels()).To(HaveKeyWithValue(model.TeamIDLabel, "7")) + }) + + It("DeleteBackup: dispatches delete-backup- with the same backup identity", func() { + stubDispatch() + + spec := provider.ServiceJobSpec(provider.ServiceJob{ + Operation: provider.JobOpDeleteBackup, + MsID: "svc-42", + Key: "bkp-7", + Image: "backup-image:1", + Command: []string{"/delete-backup"}, + Secrets: map[string]string{"BACKUP_STORE_SECRET_KEY": "shh"}, + }) + Expect(runner.Run(ctx, namespace, spec)).To(Succeed()) + + Expect(created.GetName()).To(Equal("delete-backup-bkp-7")) + Expect(created.GetName()).NotTo(Equal(provider.BackupJobName("bkp-7"))) // distinct from take-backup + Expect(created.GetLabels()).To(HaveKeyWithValue(model.ServiceIDLabel, "svc-42")) + Expect(created.GetLabels()).To(HaveKeyWithValue(model.BackupIDLabel, "bkp-7")) + }) + + It("Restore-on-create: dispatches restore-, keyed on the service", func() { + stubDispatch() + + // A provider's Create(params) builds this when params.recoverFrom is set. + spec := provider.ServiceJobSpec(provider.ServiceJob{ + Operation: provider.JobOpRestore, + MsID: "svc-42", + Key: "svc-42", // whole-service op: keyed on the service, not a backup + Image: "restore-image:1", + Command: []string{"/restore"}, + Env: map[string]string{"SOURCE_BACKUP_ID": "bkp-7"}, + Secrets: map[string]string{"SOURCE_SECRET_KEY": "shh"}, + }) + Expect(runner.Run(ctx, namespace, spec)).To(Succeed()) + + Expect(created.GetName()).To(Equal("restore-svc-42")) + Expect(created.GetLabels()).To(HaveKeyWithValue(model.ServiceIDLabel, "svc-42")) + // Restore is not backup-scoped, so it carries no backup-id label. + Expect(created.GetLabels()).NotTo(HaveKey(model.BackupIDLabel)) + }) + + It("GetBackupStatus: polls the backup job and maps it to the contract status", func() { + kube.On("Get", mock.Anything, mock.MatchedBy(func(r model.ResourceRef) bool { + return r.APIResource == model.JobAPI && r.Name == "backup-bkp-7" + })).Return(runningJob("backup-bkp-7", namespace), nil) + + st, err := runner.State(ctx, namespace, provider.BackupJobName("bkp-7")) + Expect(err).NotTo(HaveOccurred()) + Expect(provider.BackupStatusFromJob(st).Phase).To(Equal(provider.BackupPhaseRunning)) + }) + + It("DeleteBackup cleanup: removes the backup job and its owned secret", func() { + kube.On("Delete", mock.Anything, mock.MatchedBy(func(r model.ResourceRef) bool { + return r.APIResource == model.JobAPI && r.Name == "backup-bkp-7" + })).Return(nil) + kube.On("DeleteSecret", mock.Anything, namespace, "backup-bkp-7").Return(nil) + + Expect(runner.Delete(ctx, namespace, provider.BackupJobName("bkp-7"))).To(Succeed()) + }) +}) From 1a0246e0d5cd8dbc340971cdbb3f0f4df5d3aa89 Mon Sep 17 00:00:00 2001 From: utkuerol Date: Fri, 17 Jul 2026 20:23:42 +0200 Subject: [PATCH 3/4] update docs --- README.md | 94 +++++++------------------------------------------------ 1 file changed, 11 insertions(+), 83 deletions(-) diff --git a/README.md b/README.md index 3ed0f0b..c785ffb 100644 --- a/README.md +++ b/README.md @@ -53,95 +53,23 @@ server.Run() `RegisterRoutes` mounts CRUD and backup endpoints under `/api/v1/{name}`. -## Detached jobs (backup & restore) +## Detached Jobs -Backups run **detached** from the provider pod as one-shot Kubernetes Jobs. `provider.Base` exposes a `Jobs` (`client.JobRunner`) helper for dispatching and polling them, and the backup lifecycle maps onto the Job lifecycle closely. The three backup methods collapse to near one-liners: +Some operations (backups, restores, migrations) are easier to run as one-shot Kubernetes Jobs, detached from the provider pod. -```go -func (p *MyProvider) backupSpec(args model.TakeBackupArgs) client.JobSpec { - return provider.ServiceJobSpec(provider.ServiceJob{ - Operation: provider.JobOpBackup, // name: backup-, auto-labels BackupIDLabel - MsID: args.MsID, - Key: args.ID, - Image: p.backupImage, - Command: []string{"/backup"}, - // Env/Secrets are entirely yours — wire whatever your backup image reads. - Env: map[string]string{ - "BACKUP_STORE_ENDPOINT_URL": args.Config.EndpointURL, - "BACKUP_STORE_ACCESS_KEY_ID": args.Config.AccessKeyID, - }, - Secrets: map[string]string{"BACKUP_STORE_SECRET_KEY": args.Secrets.SecretKey}, - Labels: p.labels(args.MsID), - }) -} - -func (p *MyProvider) TakeBackup(ctx context.Context, args model.TakeBackupArgs) error { - return p.Jobs.Run(ctx, p.namespace(args.MsID), p.backupSpec(args)) -} - -func (p *MyProvider) GetBackupStatus(ctx context.Context, backupID string, retry model.TakeBackupArgs) (provider.BackupStatus, error) { - st, err := p.Jobs.State(ctx, p.namespace(retry.MsID), provider.BackupJobName(backupID)) - if err != nil { - return provider.BackupStatus{}, err - } - if st.Phase == client.JobFailed { - _ = p.Jobs.Replace(ctx, p.namespace(retry.MsID), p.backupSpec(retry)) - } - return provider.BackupStatusFromJob(st), nil -} - -func (p *MyProvider) DeleteBackup(ctx context.Context, args model.TakeBackupArgs) error { - return p.Jobs.Delete(ctx, p.namespace(args.MsID), provider.BackupJobName(args.ID)) -} -``` - -`ServiceJobSpec` owns the Job *conventions*: a deterministic name (`-`, so `Run` is idempotent and `State`/`Delete` resolve the same Job) and the identity labels (always `ServiceIDLabel`; backup operations also auto-stamp `BackupIDLabel` from the key). `Env` and `Secrets` pass through verbatim — each provider's image defines its own contract. Secrets are stored in the Job's owned Secret and injected via `secretKeyRef`, never appearing in the Job or Pod manifest. `Run` is idempotent on the Job name; `State` returns a phase (`pending`/`running`/`succeeded`/`failed`/`not_found`) and, on failure, folds the failed pod's logs into the reason; `Replace` is the retry path. Jobs default to a 6h deadline, a 3-retry backoff, and a 1h finished-TTL — override via `JobSpec` fields, or use `JobSpec.Customize` for anything not modelled. - -### Other detached operations - -Every detached operation is the same shape, so they all go through `provider.ServiceJob` — you supply the `Operation` prefix and execution details, and get the naming and identity-label conventions for free: - -| Operation | Job name | Auto-stamped label | -|-----------|----------|--------------------| -| `JobOpBackup` | `backup-` | `ServiceIDLabel`, `BackupIDLabel` | -| `JobOpDeleteBackup` | `delete-backup-` | `ServiceIDLabel`, `BackupIDLabel` | -| `JobOpRestore` | `restore-` | `ServiceIDLabel` | -| any provider-defined prefix | `-` | `ServiceIDLabel` | - -The operation→label policy lives in the lib, so a Job's identity labels always exist without the caller managing them; anything else you put in `Labels` rides along underneath. `ServiceIDLabel` is always stamped. The lib keeps only name helpers (`BackupJobName`, `DeleteBackupJobName`) and the status mappers (`BackupStatusFromJob` for the backup contract type, `OperationStatusFromJob` generically) — the three backup endpoints share these so they resolve the same Job. - -### Restore - -Restore is **not a standalone operation** in this contract — there is no restore endpoint. It happens **inside `Create`**: the create payload carries an optional `recoverFrom` (`model.RecoverFromBackup` or `model.RecoverFromTimestamp`, which the provider embeds in its own `CreateParams`), and when it is set the provider provisions the service *and* dispatches a detached Job to pull data from the backup store. Restore is keyed on the service ID (`restore-`) — one restore per service creation — and its progress is surfaced through the service's normal `GetStatus` (e.g. `details.ready = false` until it finishes), not a separate status endpoint. +- `client.JobRunner` (also on `provider.Base` as `Jobs`) — `Run` / `State` / `Delete` / `Replace` a one-shot Job, with an optional owned credentials Secret injected via `secretKeyRef`. +- `provider.ServiceJob` / `ServiceJobSpec` — build a `JobSpec` with a consistent name (`-`) and identity labels; `BackupStatusFromJob` / `OperationStatusFromJob` map a Job's state to a status. ```go -func (p *MyProvider) Create(ctx context.Context, params MyCreateParams) error { - // ... provision the service (StatefulSet, Service, Secret, ...) ... - - if params.RecoverFrom != nil { - spec := provider.ServiceJobSpec(provider.ServiceJob{ - Operation: provider.JobOpRestore, // name: restore- - MsID: params.ID, - Key: params.ID, - Image: p.restoreImage, - Command: []string{"/restore"}, - // recoverFrom is provider-specific; wire whatever your restore image reads. - Env: map[string]string{"SOURCE_BACKUP_ID": params.RecoverFrom.Backup.ID}, - Secrets: map[string]string{"SOURCE_SECRET_KEY": params.RecoverFrom.Backup.Secrets.SecretKey}, - Labels: p.labels(params.ID), - }) - if err := p.Jobs.Run(ctx, p.namespace(params.ID), spec); err != nil { - return err - } - } - return nil -} - -// In GetStatus, fold the restore Job's progress into the service status: -// st, _ := p.Jobs.State(ctx, ns, provider.ServiceJobName(provider.JobOpRestore, id)) -// ready := st.Phase == client.JobSucceeded || st.Phase == client.JobNotFound +spec := provider.ServiceJobSpec(provider.ServiceJob{ + Operation: provider.JobOpBackup, MsID: id, Key: backupID, + Image: img, Command: []string{"/backup"}, + Env: env, Secrets: secrets, // whatever your image reads +}) +err := p.Jobs.Run(ctx, ns, spec) ``` +See the package docs and `provider/servicejob_usage_test.go` for details. ## Configuration From fbded1901a08be94d56f44501110a9d9067266c4 Mon Sep 17 00:00:00 2001 From: utkuerol Date: Fri, 17 Jul 2026 20:25:28 +0200 Subject: [PATCH 4/4] tidy --- provider/backupjob.go | 6 ------ 1 file changed, 6 deletions(-) diff --git a/provider/backupjob.go b/provider/backupjob.go index 8ee34e6..9274b0d 100644 --- a/provider/backupjob.go +++ b/provider/backupjob.go @@ -7,12 +7,6 @@ import ( "github.com/codesphere-cloud/managed-services-lib/client" ) -// Backup is a framework contract operation (TakeBackup / GetBackupStatus / -// DeleteBackup), so the lib owns the naming and status conventions its three -// endpoints must agree on. Build the Job specs with ServiceJob using the -// JobOpBackup / JobOpDeleteBackup prefixes — both auto-stamp the backup identity -// labels; these helpers only resolve names and status. - // BackupJobName is the name of the Job that takes a backup. func BackupJobName(backupID string) string { return ServiceJobName(JobOpBackup, backupID)