diff --git a/api/dataset/v1alpha1/dataset_types.go b/api/dataset/v1alpha1/dataset_types.go index e94b0b3..981e453 100644 --- a/api/dataset/v1alpha1/dataset_types.go +++ b/api/dataset/v1alpha1/dataset_types.go @@ -24,6 +24,11 @@ import ( type DatasetStatusPhase string type DatasetType string +// AccessMode is the access granted to a namespace which references a Dataset. +// It deliberately describes a per-reference mount policy, not the access mode +// of the backing PersistentVolume. +type AccessMode string + const ( DatasetTypeGit DatasetType = "GIT" DatasetTypeS3 DatasetType = "S3" @@ -51,6 +56,11 @@ const ( _ = DatasetStatusPhaseFailed ) +const ( + AccessModeReadOnly AccessMode = "ReadOnly" + AccessModeReadWrite AccessMode = "ReadWrite" +) + type DatasetSource struct { // +kubebuilder:validation:Enum=GIT;S3;HTTP;PVC;NFS;CONDA;REFERENCE;HUGGING_FACE;MODEL_SCOPE;DATABASE;HADOOP;MANUAL // +kubebuilder:validation:XValidation:rule="self == oldSelf",message="Value is immutable" @@ -116,6 +126,8 @@ type MountOptions struct { } // DatasetSpec defines the desired state of Dataset +// +kubebuilder:validation:XValidation:rule="oldSelf == null || (has(self.shareAccess) == has(oldSelf.shareAccess) && (!has(self.shareAccess) || self.shareAccess == oldSelf.shareAccess))",message="shareAccess is immutable and cannot be added or removed" +// +kubebuilder:validation:XValidation:rule="!has(self.shareAccess) || size(self.shareAccess.rules) > 0",message="shareAccess.rules must contain at least one rule when shareAccess is configured" type DatasetSpec struct { // Share indicates whether the model is shareable with others. // When set to true, the model can be shared according to the specified selector. @@ -126,6 +138,12 @@ type DatasetSpec struct { // If Share is true and ShareToNamespaceSelector is empty, that means all namespaces can access this. // +kubebuilder:validation:Optional ShareToNamespaceSelector *metav1.LabelSelector `json:"shareToNamespaceSelector,omitempty"` + // ShareAccess assigns the effective access mode to namespaces which consume + // this Dataset through a REFERENCE Dataset. A nil value is intentionally + // distinct from an empty value: nil preserves the legacy read-only sharing + // behaviour. + // +kubebuilder:validation:Optional + ShareAccess *ShareAccess `json:"shareAccess,omitempty"` // +kubebuilder:validation:Required // source is the source of the dataset. Source DatasetSource `json:"source"` @@ -150,6 +168,22 @@ type DatasetSpec struct { DataWarmUpResources v1.ResourceRequirements `json:"resources,omitempty"` } +// ShareAccess describes the namespace based access policy for a shared Dataset. +type ShareAccess struct { + // +kubebuilder:validation:MinItems=1 + Rules []ShareAccessRule `json:"rules"` +} + +type ShareAccessRule struct { + // NamespaceSelector selects target namespaces. Empty selectors are not valid: + // an explicit rule must name at least one label requirement. + // +kubebuilder:validation:Required + // +kubebuilder:validation:XValidation:rule="size(self.matchLabels) > 0 || size(self.matchExpressions) > 0",message="namespaceSelector must not be empty" + NamespaceSelector metav1.LabelSelector `json:"namespaceSelector"` + // +kubebuilder:validation:Enum=ReadOnly;ReadWrite + AccessMode AccessMode `json:"accessMode"` +} + type VolumeClaimRef struct { // +kubebuilder:validation:Required // name is the name of the pvc. @@ -197,8 +231,27 @@ type DatasetStatus struct { PVCName string `json:"pvcName,omitempty"` // +kubebuilder:validation:Optional // readOnly indicates whether the dataset is mounted as read-only. - ReadOnly bool `json:"readOnly,omitempty"` - LastSyncTime metav1.Time `json:"lastSyncTime,omitempty"` + ReadOnly bool `json:"readOnly,omitempty"` + // mountSources is the direct-to-root source chain from the last successful + // MountPolicy resolution for a REFERENCE Dataset. It deliberately excludes this + // Dataset's own PVC; that PVC is checked from status.pvcName at verification time. + // A MountPolicy=False condition retains this binding as an anti-rebind identity pin; + // consumers must never treat it as current authorization. + MountSources []MountSource `json:"mountSources,omitempty"` + LastSyncTime metav1.Time `json:"lastSyncTime,omitempty"` +} + +// MountSource pins every Dataset and storage object used by a reference mount. +// Consumers must verify these identities before treating ReadOnly=false as a +// writable mount. +type MountSource struct { + Namespace string `json:"namespace"` + Name string `json:"name"` + UID string `json:"uid"` + PVCName string `json:"pvcName"` + PVCUID string `json:"pvcUID"` + PVName string `json:"pvName"` + PVUID string `json:"pvUID"` } // Dataset is the Schema for the datasets API diff --git a/api/dataset/v1alpha1/dataset_types_envtest_test.go b/api/dataset/v1alpha1/dataset_types_envtest_test.go new file mode 100644 index 0000000..46cd1e5 --- /dev/null +++ b/api/dataset/v1alpha1/dataset_types_envtest_test.go @@ -0,0 +1,80 @@ +package v1alpha1_test + +import ( + "context" + "os" + "path/filepath" + goruntime "runtime" + "testing" + + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/envtest" + + datasetv1alpha1 "github.com/BaizeAI/dataset/api/dataset/v1alpha1" +) + +func TestShareAccessCRDValidation(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("envtest process cleanup is unsupported on Windows; run this test on Linux CI") + } + if os.Getenv("KUBEBUILDER_ASSETS") == "" { + t.Skip("KUBEBUILDER_ASSETS is required for envtest") + } + _, file, _, ok := goruntime.Caller(0) + require.True(t, ok) + crdPath := filepath.Join(filepath.Dir(file), "..", "..", "..", "config", "crd", "bases") + testEnv := &envtest.Environment{CRDDirectoryPaths: []string{crdPath}, ErrorIfCRDPathMissing: true} + cfg, err := testEnv.Start() + require.NoError(t, err) + defer func() { require.NoError(t, testEnv.Stop()) }() + + scheme := runtime.NewScheme() + require.NoError(t, datasetv1alpha1.AddToScheme(scheme)) + apiClient, err := client.New(cfg, client.Options{Scheme: scheme}) + require.NoError(t, err) + ctx := context.Background() + + validRule := datasetv1alpha1.ShareAccessRule{ + NamespaceSelector: metav1.LabelSelector{MatchLabels: map[string]string{"workspace": "one"}}, + AccessMode: datasetv1alpha1.AccessModeReadOnly, + } + newDataset := func(name string) *datasetv1alpha1.Dataset { + return &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}, + Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeManual, URI: "manual://"}}, + } + } + + t.Run("rejects explicit empty rules even before sharing is enabled", func(t *testing.T) { + ds := newDataset("empty-rules") + ds.Spec.ShareAccess = &datasetv1alpha1.ShareAccess{} + require.Error(t, apiClient.Create(ctx, ds)) + }) + t.Run("rejects empty namespace selector", func(t *testing.T) { + ds := newDataset("empty-selector") + ds.Spec.ShareAccess = &datasetv1alpha1.ShareAccess{Rules: []datasetv1alpha1.ShareAccessRule{{AccessMode: datasetv1alpha1.AccessModeReadOnly}}} + require.Error(t, apiClient.Create(ctx, ds)) + }) + t.Run("allows a nonempty preconfigured policy and later enabling sharing", func(t *testing.T) { + ds := newDataset("preconfigured") + ds.Spec.ShareAccess = &datasetv1alpha1.ShareAccess{Rules: []datasetv1alpha1.ShareAccessRule{validRule}} + require.NoError(t, apiClient.Create(ctx, ds)) + ds.Spec.Share = true + require.NoError(t, apiClient.Update(ctx, ds)) + }) + t.Run("rejects adding or changing a policy after creation", func(t *testing.T) { + missing := newDataset("missing-policy") + require.NoError(t, apiClient.Create(ctx, missing)) + missing.Spec.ShareAccess = &datasetv1alpha1.ShareAccess{Rules: []datasetv1alpha1.ShareAccessRule{validRule}} + require.Error(t, apiClient.Update(ctx, missing)) + + immutable := newDataset("immutable-policy") + immutable.Spec.ShareAccess = &datasetv1alpha1.ShareAccess{Rules: []datasetv1alpha1.ShareAccessRule{validRule}} + require.NoError(t, apiClient.Create(ctx, immutable)) + immutable.Spec.ShareAccess.Rules[0].AccessMode = datasetv1alpha1.AccessModeReadWrite + require.Error(t, apiClient.Update(ctx, immutable)) + }) +} diff --git a/api/dataset/v1alpha1/zz_generated.deepcopy.go b/api/dataset/v1alpha1/zz_generated.deepcopy.go index f7b5230..7255c4e 100644 --- a/api/dataset/v1alpha1/zz_generated.deepcopy.go +++ b/api/dataset/v1alpha1/zz_generated.deepcopy.go @@ -131,6 +131,11 @@ func (in *DatasetSpec) DeepCopyInto(out *DatasetSpec) { *out = new(v1.LabelSelector) (*in).DeepCopyInto(*out) } + if in.ShareAccess != nil { + in, out := &in.ShareAccess, &out.ShareAccess + *out = new(ShareAccess) + (*in).DeepCopyInto(*out) + } in.Source.DeepCopyInto(&out.Source) out.MountOptions = in.MountOptions in.VolumeClaimTemplate.DeepCopyInto(&out.VolumeClaimTemplate) @@ -139,6 +144,7 @@ func (in *DatasetSpec) DeepCopyInto(out *DatasetSpec) { *out = new(VolumeClaimRef) **out = **in } + in.DataWarmUpResources.DeepCopyInto(&out.DataWarmUpResources) } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DatasetSpec. @@ -168,6 +174,11 @@ func (in *DatasetStatus) DeepCopyInto(out *DatasetStatus) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.MountSources != nil { + in, out := &in.MountSources, &out.MountSources + *out = make([]MountSource, len(*in)) + copy(*out, *in) + } in.LastSyncTime.DeepCopyInto(&out.LastSyncTime) } @@ -196,6 +207,59 @@ func (in *MountOptions) DeepCopy() *MountOptions { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *MountSource) DeepCopyInto(out *MountSource) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new MountSource. +func (in *MountSource) DeepCopy() *MountSource { + if in == nil { + return nil + } + out := new(MountSource) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ShareAccess) DeepCopyInto(out *ShareAccess) { + *out = *in + if in.Rules != nil { + in, out := &in.Rules, &out.Rules + *out = make([]ShareAccessRule, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ShareAccess. +func (in *ShareAccess) DeepCopy() *ShareAccess { + if in == nil { + return nil + } + out := new(ShareAccess) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ShareAccessRule) DeepCopyInto(out *ShareAccessRule) { + *out = *in + in.NamespaceSelector.DeepCopyInto(&out.NamespaceSelector) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ShareAccessRule. +func (in *ShareAccessRule) DeepCopy() *ShareAccessRule { + if in == nil { + return nil + } + out := new(ShareAccessRule) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *VolumeClaimRef) DeepCopyInto(out *VolumeClaimRef) { *out = *in diff --git a/config/crd/bases/dataset.baizeai.io_datasets.yaml b/config/crd/bases/dataset.baizeai.io_datasets.yaml index ce832d0..c6ee2d4 100644 --- a/config/crd/bases/dataset.baizeai.io_datasets.yaml +++ b/config/crd/bases/dataset.baizeai.io_datasets.yaml @@ -158,6 +158,86 @@ spec: Share indicates whether the model is shareable with others. When set to true, the model can be shared according to the specified selector. type: boolean + shareAccess: + description: |- + ShareAccess assigns the effective access mode to namespaces which consume + this Dataset through a REFERENCE Dataset. A nil value is intentionally + distinct from an empty value: nil preserves the legacy read-only sharing + behaviour. + properties: + rules: + items: + properties: + accessMode: + description: |- + AccessMode is the access granted to a namespace which references a Dataset. + It deliberately describes a per-reference mount policy, not the access mode + of the backing PersistentVolume. + enum: + - ReadOnly + - ReadWrite + type: string + namespaceSelector: + description: |- + NamespaceSelector selects target namespaces. Empty selectors are not valid: + an explicit rule must name at least one label requirement. + properties: + matchExpressions: + description: matchExpressions is a list of label selector + requirements. The requirements are ANDed. + items: + description: |- + A label selector requirement is a selector that contains values, a key, and an operator that + relates the key and values. + properties: + key: + description: key is the label key that the selector + applies to. + type: string + operator: + description: |- + operator represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: |- + values is an array of string values. If the operator is In or NotIn, + the values array must be non-empty. If the operator is Exists or DoesNotExist, + the values array must be empty. This array is replaced during a strategic + merge patch. + items: + type: string + type: array + x-kubernetes-list-type: atomic + required: + - key + - operator + type: object + type: array + x-kubernetes-list-type: atomic + matchLabels: + additionalProperties: + type: string + description: |- + matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels + map is equivalent to an element of matchExpressions, whose key field is "key", the + operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + x-kubernetes-map-type: atomic + x-kubernetes-validations: + - message: namespaceSelector must not be empty + rule: size(self.matchLabels) > 0 || size(self.matchExpressions) + > 0 + required: + - accessMode + - namespaceSelector + type: object + minItems: 1 + type: array + required: + - rules + type: object shareToNamespaceSelector: description: |- ShareToNamespaceSelector defines a label selector to specify the namespaces @@ -401,7 +481,7 @@ spec: resources: description: |- resources represents the minimum resources the volume should have. - If RecoverVolumeExpansionFailure feature is enabled users are allowed to specify resource requirements + Users are allowed to specify resource requirements that are lower than previous value but must still be higher than capacity recorded in the status field of the claim. More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#resources @@ -558,8 +638,7 @@ spec: for the purpose it was designed. For example - a controller that\nonly is responsible for resizing capacity of the volume, should ignore PVC updates that change other valid\nresources - associated with PVC.\n\nThis is an alpha field and requires - enabling RecoverVolumeExpansionFailure feature." + associated with PVC." type: object x-kubernetes-map-type: granular allocatedResources: @@ -590,8 +669,7 @@ spec: for the purpose it was designed. For example - a controller that\nonly is responsible for resizing capacity of the volume, should ignore PVC updates that change other valid\nresources - associated with PVC.\n\nThis is an alpha field and requires - enabling RecoverVolumeExpansionFailure feature." + associated with PVC." type: object capacity: additionalProperties: @@ -690,6 +768,13 @@ spec: required: - source type: object + x-kubernetes-validations: + - message: shareAccess is immutable and cannot be added or removed + rule: oldSelf == null || (has(self.shareAccess) == has(oldSelf.shareAccess) + && (!has(self.shareAccess) || self.shareAccess == oldSelf.shareAccess)) + - message: shareAccess.rules must contain at least one rule when shareAccess + is configured + rule: '!has(self.shareAccess) || size(self.shareAccess.rules) > 0' status: description: DatasetStatus defines the observed state of Dataset properties: @@ -762,6 +847,43 @@ spec: lastSyncTime: format: date-time type: string + mountSources: + description: |- + mountSources is the direct-to-root source chain from the last successful + MountPolicy resolution for a REFERENCE Dataset. It deliberately excludes this + Dataset's own PVC; that PVC is checked from status.pvcName at verification time. + A MountPolicy=False condition retains this binding as an anti-rebind identity pin; + consumers must never treat it as current authorization. + items: + description: |- + MountSource pins every Dataset and storage object used by a reference mount. + Consumers must verify these identities before treating ReadOnly=false as a + writable mount. + properties: + name: + type: string + namespace: + type: string + pvName: + type: string + pvUID: + type: string + pvcName: + type: string + pvcUID: + type: string + uid: + type: string + required: + - name + - namespace + - pvName + - pvUID + - pvcName + - pvcUID + - uid + type: object + type: array phase: default: PENDING type: string diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index dd78d52..bff8671 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -29,6 +29,7 @@ rules: resources: - persistentvolumes verbs: + - create - delete - get - list diff --git a/docs/dataset-repository-share-access-implementation.md b/docs/dataset-repository-share-access-implementation.md new file mode 100644 index 0000000..10a272a --- /dev/null +++ b/docs/dataset-repository-share-access-implementation.md @@ -0,0 +1,209 @@ +# Dataset Repository: Target-Workspace Share-Access Implementation Checklist + +This document covers only the changes required in the `D:\github_repositories\dataset` repository (github.com/BaizeAI/dataset). It is released alongside baize's management API and Pod-mount consumers; it does not cover baize protobufs, UI, or training, Notebook, and inference code. + +## 1. Goals and Boundaries + +When Dataset A enables sharing, access is granted according to the workspace bound to the **target namespace in which the referencer resides**: + +- Targets such as workspaces 1 and 3 receive read-only access. +- Targets such as workspaces 2 and 4 receive read-write access. +- Targets that do not match are denied when creating or using a reference. +- A's own direct use does not automatically become read-only merely because A is shared externally as read-only. + +When `REFERENCE` Dataset B references A, it does not copy file data. The Controller prepares separate PVC/PV objects for B, while their underlying storage continues to point to A. B's `status.readOnly` is B's effective permission in its target namespace. + +Do not add a global `spec.readOnly`. `ReadWriteMany` on the underlying PV/PVC represents the storage's concurrent-mount capability; it is not a business-level read/write permission assigned by namespace. + +## 2. API and CRD + +Modify `api/dataset/v1alpha1/dataset_types.go`, then regenerate `zz_generated.deepcopy.go` and the CRD. + +### 2.1 `spec.shareAccess` + +Add the following optional **pointer** field to `DatasetSpec`. The presence or absence of the field must remain semantically meaningful: + +```go +ShareAccess *ShareAccess `json:"shareAccess,omitempty"` +``` + +Its value types are: + +```go +type ShareAccess struct { + Rules []ShareAccessRule `json:"rules"` +} + +type ShareAccessRule struct { + NamespaceSelector metav1.LabelSelector `json:"namespaceSelector"` + // +kubebuilder:validation:Enum=ReadOnly;ReadWrite + AccessMode AccessMode `json:"accessMode"` +} + +// AccessMode permits only read-only or read-write. The kubebuilder marker enforces +// the enum at the CRD schema layer and rejects invalid values. +type AccessMode string + +const ( + AccessModeReadOnly AccessMode = "ReadOnly" + AccessModeReadWrite AccessMode = "ReadWrite" +) +``` + +Constraints: + +- `accessMode` may only be `ReadOnly` or `ReadWrite`; enforce this at the CRD schema layer with the kubebuilder Enum marker on `AccessMode`. +- `namespaceSelector` is required and must not be an empty selector. +- When `share=true` and `shareAccess` is present, `rules` must contain at least one rule. +- `shareAccess` is immutable after creation: it cannot be changed, deleted, or populated later on historical objects. +- When `share=false`, a non-empty `shareAccess` may be preconfigured at creation time. Preserve it and grant access according to its rules when sharing is later enabled. If it is absent at creation, it cannot be populated later; after sharing is enabled, continue to use the historical read-only model. +- `shareAccess` being absent must be distinguished from an explicitly empty object or empty `rules`. Only absence selects the backward-compatible model; it does not mean read-write. Historical shared Datasets are always read-only within the range of the original `shareToNamespaceSelector`. An explicit empty configuration must be rejected when `share=true`; it must never fall back to the historical model. +- `shareToNamespaceSelector` remains the overall sharing boundary. Actual authorization must pass both it and `shareAccess.rules`. + +Use kubebuilder validation/CEL at the CRD layer to validate field presence and immutability. Immutability rules must be based on `has(self.shareAccess)` / `has(oldSelf.shareAccess)`; do not collapse the pointer field into a value type. The Controller must also defensively reject invalid CRs submitted directly. + +### 2.2 `status.mountSources` + +Add the following field to `DatasetStatus`. For a `REFERENCE` Dataset, it is the complete reference chain ordered from its direct source to its ultimate root source, and **does not include itself**: for `C -> B -> A`, `C.status.mountSources` is `[B, A]`. Each item records the corresponding Dataset and its actual PVC/PV identities. C's own PVC is found via `C.status.PVCName`, and its actual UID, owner, and PV binding are verified in `Bindings`/`Verify`. This field is not written for non-`REFERENCE` Datasets. + +```go +MountSources []MountSource `json:"mountSources,omitempty"` +``` + +The item type binds the authorization policy to the actual storage: + +```go +type MountSource struct { + Namespace string `json:"namespace"` + Name string `json:"name"` + UID string `json:"uid"` + PVCName string `json:"pvcName"` + PVCUID string `json:"pvcUID"` + PVName string `json:"pvName"` + PVUID string `json:"pvUID"` +} +``` + +Keep `status.readOnly`; it represents the current Dataset's effective mount permission, not A's global policy for all namespaces. When a reference Dataset succeeds, write `status.readOnly`, the complete `status.mountSources`, and a `MountPolicy=True` condition carrying the current Dataset's `observedGeneration` in the same status update. On failure, write `MountPolicy=False`; do not allow an old `readOnly=false` to continue being treated as authorization. + +Keep `ReadOnly` as `omitempty`: when serialized to JSON, `readOnly=false` is omitted, so unstructured/JSON consumers cannot distinguish “write allowed” from “not yet computed.” `MountPolicy=True` with an `observedGeneration` equal to the current generation is therefore the authoritative signal that the permission has been computed successfully. A missing or stale condition must never permit writes. + +## 3. Shared Authorization Resolver + +Add `pkg/mountpolicy`, shared by the Controller and baize, so the two sides do not interpret the rules independently. + +It should provide at least the following capabilities: + +| Function | Responsibility | +| --- | --- | +| `Validate` | Validate rule modes, selectors, and sharing configuration | +| `Grant` | Determine Denied / ReadOnly / ReadWrite for a source Dataset and target namespace | +| `Resolve` | Resolve the reference chain, cycles, and depth, and calculate the effective read-only result | +| `Bindings` | Read and verify the UIDs of the Dataset, PVC, and PV, along with the actual PV source | +| `Verify` | Let consumers confirm that a reference Dataset is prepared; that `MountPolicy=True`; that the condition's `observedGeneration` equals the current generation; that the full source chain and PVC/PV identities still match; and that resolving the current chain again still grants access | +| `ProtectedPVC` | Determine whether a PVC is the protected volume of a reference Dataset, preventing alias-based bypasses | + +### 3.1 Single-Level Rules + +`Grant(source, targetNamespace)` follows these rules: + +1. Deny if the source is not shared or is being deleted. +2. If `shareToNamespaceSelector` is nil or structurally empty, preserve the existing semantics: the overall sharing range is every namespace. Deny only when the selector is non-empty and the target does not match. +3. If `shareAccess` is absent, return read-only within the overall range for backward compatibility. Thus, a historical shared Dataset whose overall range is every namespace is read-only in every namespace. +4. If no new rule matches, deny. A nil or empty `shareToNamespaceSelector` does not relax `shareAccess.rules`; new rules still authorize only matching targets. +5. When multiple rules match, read-only takes precedence; return read-write only if all matching rules are read-write. +6. Fail closed for any selector, namespace, or object-read error; never fall back to read-write. + +Workspace labels on the target namespace must originate from a trusted platform identity. Do not trust a workspace ID supplied by the caller. + +Before generating a Pod, consumers must call `Verify`, rather than decide writability from `status.readOnly` alone. `Verify` must fail closed for any missing, stale, identity-mismatched, chain-re-resolution-failed, or unauthorized result. Only after successful verification may a consumer interpret `status.readOnly=false` as write permission. + +### 3.2 Multi-Level References + +Support `C -> B -> A`. The recommended maximum depth is 32; reject self-references, cycles, deleted sources, missing sources, and `REFERENCE + volumeClaimRef`. + +For C's final namespace: + +- Every direct reference edge must be authorized. +- Every upstream A/B must also authorize the final namespace. +- If any edge is read-only, the final result is read-only. +- If any edge is denied, reject the entire reference. + +Do not treat an upstream legacy `status.readOnly=false` as authorization; resolve the current chain again. + +## 4. Dataset Controller + +Modify `internal/controller/dataset/dataset_controller.go`. + +### 4.1 Validation and Reconciliation Order + +For `REFERENCE`: + +```text +Validate sharing rules and the reference chain +-> Prepare/confirm B's PVC and PV +-> Calculate MountPolicy, status.readOnly, and MountSources +-> Write the status condition +``` + +Requirements: + +- Remove the existing logic that unconditionally writes `status.readOnly=true`. +- Even when the PVC is Ready, continue MountPolicy reconciliation; do not return early. +- If existing `MountSources` do not match the currently resolved sources, deny authorization; do not rebind to a newly created A with the same name. +- Set `MountPolicy=False` when MountPolicy fails. A PVC-preparation failure must also set `MountPolicy=False` (or calculate authorization before PVC preparation); no previous `MountPolicy=True` may remain. A condition's `observedGeneration` reflects only this object's spec generation, so upstream policy changes do not make it stale; do not rely on chain re-resolution in `Verify` as an implicit safeguard. +- A `REFERENCE` Dataset must keep retrying periodically after entering Failed: change the `Reconcile()` branch that currently returns `resOk` immediately for `Failed` so that a REFERENCE returns `res30sec` (or an equivalent bounded `RequeueAfter`). An error from any reconciler interrupts later steps, so recovery cannot rely solely on watch events. +- Watch Dataset and Namespace changes and requeue every direct and transitive reference. Periodic reconciliation is a fallback for lost watches and source recovery. +- An authorization-resolution failure must not block deletion through the finalizer path. + +### 4.2 PVC/PV Lifecycle + +B's reference PV is a copy of A's PV configuration. At creation it must: + +- Clear `ResourceVersion`, UID, creation timestamp, managed fields, status, and `claimRef`. +- Record the source Dataset UID and source PV UID. +- Use B as the owner of B's PVC/PV. +- Fix the cloned PV's `persistentVolumeReclaimPolicy` to `Retain`, preventing deletion of B from deleting the shared source data. +- Verify B's actual PVC/PV UIDs, PV `claimRef`, and underlying `PersistentVolumeSource` match the recorded values. +- Avoid a fixed-length UID slice when naming cloned PVs; use a safe truncation helper or the full UID, so short fake-client UIDs cannot panic. + +Deleting B may reclaim B's objects but must not delete A's underlying data. Do not recreate A's storage in response to a permission change. + +### 4.3 Preventing PVC Alias Bypasses + +Reject an ordinary `PVC` Dataset or `volumeClaimRef` when it points to the PVC of reference Dataset B, and instruct users to use the original Dataset/Model reference path. `ProtectedPVC` must use the live Dataset list; do not allow access when the query fails or associations conflict. + +## 5. Tests + +Add at least the following tests: + +- Read-only groups, read-write groups, unmatched namespaces, and historical objects with `shareAccess` absent. +- A nil or structurally empty `shareToNamespaceSelector`: historical objects are read-only in all namespaces; objects using new rules still authorize only namespaces matching `shareAccess.rules`. +- Valid `shareAccess` preconfigured when created with `share=false`, then authorization according to the original rules after sharing is enabled; also distinguish an absent field, an explicitly empty object, and empty `rules`. +- When JSON/unstructured representation omits `status.ReadOnly=false`, `Verify` may return writable only with `MountPolicy=True` and matching `observedGeneration`. +- Read-only wins where overlapping rules match the same namespace. +- Multi-level references: upstream read-only, upstream not authorizing the final namespace, fully read-write chains, cycles, and excessive depth. +- Effective permission is reconciled again after a Ready PVC when namespace workspace labels or source policy changes. +- A REFERENCE that enters Failed due to a temporary source/policy error automatically recovers through periodic requeue even without an additional watch event. +- A PVC-preparation failure (such as a source without a PVC or a PVC not bound to a PV) must set `MountPolicy=False` and must not leave an old condition behind. +- Deny mounting when the source Dataset/PVC/PV is recreated or its UID does not match. +- For `C -> B -> A`, `C.status.mountSources` must be exactly `[B, A]` and exclude C. Validate C's own volume with `status.PVCName` and actual UID/owner/PV binding. `Verify` must reject a same-name recreation of any chain member or a change to PVC/PV UIDs or the actual PV source. +- Reject wrapping B's PVC in an ordinary PVC Dataset or `volumeClaimRef`. +- The cloned PV retains `Retain` and source UIDs; short fake-client UIDs do not panic during PV-name slicing. +- CRD schema/CEL: invalid modes, empty selectors, empty rules, and creation-time modification, deletion, or later population of `shareAccess`. + +The fake client may cover Controller branches; CRD/CEL should also use the Kubernetes API validator or envtest. Before the final release, validate in a real NFS/CSI environment: read-only workspaces cannot write, read-write workspaces can write, and deleting B does not affect A's data. + +## 6. Release Contract with baize + +baize must depend on the module version published from this repository rather than rely indefinitely on a local `go.work` reference to `../dataset`. + +Release order: + +1. Release the Dataset API, CRD, and Controller; first complete MountPolicy/source-identity reconciliation for historical reference objects. +2. Release baize's management API and mount consumers. Every Pod-generation entry point must use the same `Verify` result before generating volumes and `volumeMounts`; it must not read only `status.readOnly`. +3. After validating version compatibility, historical template migration, and real storage, enable configuration of read-write workspaces. + +Do not open the new read-write rules until the Controller and CRD are fully upgraded, so an old Controller cannot hard-code every reference as read-only. + +Rollback boundary: once read-write rules and running read-write Pods exist, a lossless rollback is impossible—the old Controller will change references back to read-only, but mounts of already running Pods remain unchanged. For rollback, first disable the read-write workspace configuration entry point, then prioritize a forward fix and reconcile workload recreation. Do not roll back by deleting `shareAccess` or recreating the underlying storage. diff --git a/internal/controller/dataset/dataset_controller.go b/internal/controller/dataset/dataset_controller.go index 2aa5464..baec41f 100644 --- a/internal/controller/dataset/dataset_controller.go +++ b/internal/controller/dataset/dataset_controller.go @@ -26,7 +26,7 @@ import ( "time" "github.com/BaizeAI/dataset/pkg/kubeutils" - "k8s.io/apimachinery/pkg/labels" + "github.com/BaizeAI/dataset/pkg/mountpolicy" "github.com/samber/lo" batchv1 "k8s.io/api/batch/v1" @@ -44,7 +44,12 @@ import ( "k8s.io/apimachinery/pkg/runtime" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" datasetv1alpha1 "github.com/BaizeAI/dataset/api/dataset/v1alpha1" ) @@ -53,11 +58,12 @@ const ( datasetFinalizer = "dataset-controller" keepConditions = 5 - condTypeConfig = "Config" - condTypePVC = "PVC" - condTypeJobStatus = "JobStatus" - condTypeJob = "Job" - condTypeConfigMap = "ConfigMap" + condTypeConfig = "Config" + condTypePVC = "PVC" + condTypeJobStatus = "JobStatus" + condTypeJob = "Job" + condTypeConfigMap = "ConfigMap" + condTypeMountPolicy = mountpolicy.MountPolicyCondition nfsPersistentVolumeTemplate = ` apiVersion: v1 @@ -91,7 +97,7 @@ type reconciler struct { //+kubebuilder:rbac:groups=dataset.baizeai.io,resources=datasets,verbs=get;list;watch;create;update;patch;delete //+kubebuilder:rbac:groups=dataset.baizeai.io,resources=datasets/status,verbs=get;update;patch //+kubebuilder:rbac:groups=dataset.baizeai.io,resources=datasets/finalizers,verbs=update -//+kubebuilder:rbac:groups="",resources=persistentvolumes,verbs=get;list;watch;delete +//+kubebuilder:rbac:groups="",resources=persistentvolumes,verbs=get;list;watch;create;delete //+kubebuilder:rbac:groups="",resources=persistentvolumeclaims,verbs=get;list;watch;create;update;patch;delete //+kubebuilder:rbac:groups="",resources=namespaces,verbs=get;list;watch @@ -119,6 +125,7 @@ func (r *DatasetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct {typ: condTypeConfig, rec: r.validate}, {typ: "", rec: r.reconcileFinalizer}, {typ: condTypePVC, rec: r.reconcilePVC}, + {typ: condTypeMountPolicy, rec: r.reconcileMountPolicy}, {typ: condTypeConfigMap, rec: r.reconcileConfigMap}, {typ: condTypeJob, rec: r.reconcileJob}, {typ: condTypeJobStatus, rec: r.reconcileJobStatus}, @@ -128,7 +135,19 @@ func (r *DatasetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct for _, rr := range reconcilers { log.Debugf("start reconciling dataset for %s/%s: %+v...", ds.Namespace, ds.Name, rr) err := rr.rec(ctx, ds) - ds.Status.Conditions = kubeutils.SetCondition(ds.Status.Conditions, rr.typ, err) + if rr.typ == condTypeMountPolicy { + if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeReference { + r.setMountPolicyCondition(ds, err) + } + } else { + ds.Status.Conditions = kubeutils.SetCondition(ds.Status.Conditions, rr.typ, err) + // A failed step before MountPolicy must invalidate an older success + // condition; otherwise stale status.ReadOnly=false could be mistaken + // for authorization by an older consumer. + if err != nil && ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeReference && !kubeutils.IsDeleted(ds) { + r.setMountPolicyCondition(ds, err) + } + } if err != nil { log.Errorf("error reconciling dataset for %s/%s: %v", ds.Namespace, ds.Name, err) break @@ -152,6 +171,12 @@ func (r *DatasetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct } } + // REFERENCE datasets periodically re-resolve their source chain. Watches + // make normal updates prompt, while this is the recovery path for missed + // events and for a reference which entered Failed before its source existed. + if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeReference { + return res30sec, nil + } switch ds.Status.Phase { case datasetv1alpha1.DatasetStatusPhaseReady, datasetv1alpha1.DatasetStatusPhaseFailed: return resOk, nil @@ -228,18 +253,11 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha switch ds.Spec.Source.Type { case datasetv1alpha1.DatasetTypeReference: if kubeutils.IsDeleted(ds) { - // Enhanced cleanup for reference datasets - also handle retained PVs if config.IsCascadingDeletionEnabled() { - // Find and delete the associated retained PV if err := r.cleanupRetainedPV(ctx, ds); err != nil { - log.Errorf("Failed to cleanup retained PV for dataset %s/%s: %v", ds.Namespace, ds.Name, err) - // Don't fail the deletion process if PV cleanup fails + log.Errorf("cleanup retained pv for reference dataset %s/%s: %v", ds.Namespace, ds.Name, err) } } - // OwnerReference 会将其自动回收，这里不做额外 Delete - return nil - } - if kubeutils.IsConditionReady(ds.Status.Conditions, condTypePVC) { return nil } srcDs, err := r.getSourceDataset(ctx, ds) @@ -249,50 +267,62 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha if srcDs.Status.PVCName == "" { return fmt.Errorf("source dataset %s/%s has no pvc", srcDs.Namespace, srcDs.Name) } - // 先获取 source dataset 的 pvc - pvc := &corev1.PersistentVolumeClaim{} - err = r.Get(ctx, client.ObjectKey{Namespace: srcDs.Namespace, Name: srcDs.Status.PVCName}, pvc) - if err != nil { - return fmt.Errorf("get pvc %s/%s for source dataset %s/%s error: %v", - srcDs.Namespace, srcDs.Status.PVCName, - srcDs.Namespace, srcDs.Name, err) + sourcePVC := &corev1.PersistentVolumeClaim{} + if err := r.Get(ctx, client.ObjectKey{Namespace: srcDs.Namespace, Name: srcDs.Status.PVCName}, sourcePVC); err != nil { + return fmt.Errorf("get source pvc %s/%s: %w", srcDs.Namespace, srcDs.Status.PVCName, err) } - if pvc.Spec.VolumeName == "" { - return fmt.Errorf("pvc %s/%s has no volume", pvc.Namespace, pvc.Name) + if sourcePVC.Status.Phase != corev1.ClaimBound || sourcePVC.Spec.VolumeName == "" { + return fmt.Errorf("source pvc %s/%s is not bound", sourcePVC.Namespace, sourcePVC.Name) } - // 再获取 source dataset pvc 对应的 pv - pv := &corev1.PersistentVolume{} - err = r.Get(ctx, client.ObjectKey{Name: pvc.Spec.VolumeName}, pv) - if err != nil { - return fmt.Errorf("get pv %s for source dataset %s/%s error: %v", - pvc.Spec.VolumeName, srcDs.Namespace, srcDs.Name, err) - } - // 克隆一个新的 pv 给当前 ds - newPv := pv.DeepCopy() - newPv.OwnerReferences = datasetOwnerRef(ds) - newPv.Name = fmt.Sprintf("dataset-%s-%s-%s", ds.Namespace, ds.Name, ds.UID[:12]) - if newPv.Labels == nil { - newPv.Labels = make(map[string]string) - } - newPv.Labels[constants.DatasetNameLabel] = ds.Name - newPv.ResourceVersion = "" - newPv.Spec.ClaimRef = nil - // 保留策略改为 Retain - newPv.Spec.PersistentVolumeReclaimPolicy = corev1.PersistentVolumeReclaimRetain - if err := r.Get(ctx, client.ObjectKey{Name: newPv.Name}, pv); err != nil { + sourcePV := &corev1.PersistentVolume{} + if err := r.Get(ctx, client.ObjectKey{Name: sourcePVC.Spec.VolumeName}, sourcePV); err != nil { + return fmt.Errorf("get source pv %s: %w", sourcePVC.Spec.VolumeName, err) + } + if sourcePV.Spec.ClaimRef == nil || sourcePV.Spec.ClaimRef.Namespace != sourcePVC.Namespace || sourcePV.Spec.ClaimRef.Name != sourcePVC.Name || sourcePV.Spec.ClaimRef.UID != sourcePVC.UID { + return fmt.Errorf("source pv %s is not bound to pvc %s/%s", sourcePV.Name, sourcePVC.Namespace, sourcePVC.Name) + } + + pvName := referencePVName(ds) + existingPV := &corev1.PersistentVolume{} + if err := r.Get(ctx, client.ObjectKey{Name: pvName}, existingPV); err != nil { if !k8serrors.IsNotFound(err) { return err } - if err := r.Create(ctx, newPv); err != nil { + newPV := sourcePV.DeepCopy() + newPV.ObjectMeta = metav1.ObjectMeta{ + Name: pvName, + Labels: copyStringMap(sourcePV.Labels), + Annotations: copyStringMap(sourcePV.Annotations), + OwnerReferences: datasetOwnerRef(ds), + } + if newPV.Labels == nil { + newPV.Labels = map[string]string{} + } + newPV.Labels[constants.DatasetNameLabel] = ds.Name + if newPV.Annotations == nil { + newPV.Annotations = map[string]string{} + } + newPV.Annotations[mountpolicy.SourceDatasetUIDAnnotation] = string(srcDs.UID) + newPV.Annotations[mountpolicy.SourcePVCUIDAnnotation] = string(sourcePVC.UID) + newPV.Annotations[mountpolicy.SourcePVUIDAnnotation] = string(sourcePV.UID) + newPV.Spec.ClaimRef = nil + newPV.Spec.PersistentVolumeReclaimPolicy = corev1.PersistentVolumeReclaimRetain + newPV.Status = corev1.PersistentVolumeStatus{} + if err := r.Create(ctx, newPV); err != nil { return err } + } else { + if existingPV.Labels[constants.DatasetNameLabel] != ds.Name || + existingPV.Annotations[mountpolicy.SourceDatasetUIDAnnotation] != string(srcDs.UID) || + existingPV.Annotations[mountpolicy.SourcePVCUIDAnnotation] != string(sourcePVC.UID) || + existingPV.Annotations[mountpolicy.SourcePVUIDAnnotation] != string(sourcePV.UID) || + !reflect.DeepEqual(existingPV.Spec.PersistentVolumeSource, sourcePV.Spec.PersistentVolumeSource) { + return fmt.Errorf("reference pv %s does not match its source binding", existingPV.Name) + } } - spec = pvc.Spec.DeepCopy() - spec.VolumeName = newPv.Name - - // 标记当前 dataset 状态 + spec = sourcePVC.Spec.DeepCopy() + spec.VolumeName = pvName ds.Status.LastSucceedRound = ds.Spec.DataSyncRound - ds.Status.ReadOnly = true case datasetv1alpha1.DatasetTypePVC: u, err := url.Parse(ds.Spec.Source.URI) @@ -501,8 +531,15 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha } func (r *DatasetReconciler) reconcileClaimPVC(ctx context.Context, ds *datasetv1alpha1.Dataset) error { + protected, err := mountpolicy.ProtectedPVC(ctx, r.Client, ds.Namespace, ds.Spec.VolumeClaimRef.Name) + if err != nil { + return err + } + if protected { + return fmt.Errorf("pvc %s/%s is managed by a REFERENCE dataset; use the dataset reference instead", ds.Namespace, ds.Spec.VolumeClaimRef.Name) + } var pvc corev1.PersistentVolumeClaim - err := r.Get(ctx, client.ObjectKey{Namespace: ds.Namespace, Name: ds.Spec.VolumeClaimRef.Name}, &pvc) + err = r.Get(ctx, client.ObjectKey{Namespace: ds.Namespace, Name: ds.Spec.VolumeClaimRef.Name}, &pvc) if err != nil { return fmt.Errorf("get pvc %s/%s for dataset %s/%s error: %v", ds.Namespace, ds.Spec.VolumeClaimRef.Name, ds.Namespace, ds.Name, err) } @@ -911,6 +948,89 @@ func (r *DatasetReconciler) reconcileJobStatus(ctx context.Context, ds *datasetv return nil } +func (r *DatasetReconciler) reconcileMountPolicy(ctx context.Context, ds *datasetv1alpha1.Dataset) error { + if ds.Spec.Source.Type != datasetv1alpha1.DatasetTypeReference || kubeutils.IsDeleted(ds) { + return nil + } + resolution, err := mountpolicy.Resolve(ctx, r.Client, ds) + if err != nil { + return err + } + bindings, err := mountpolicy.Bindings(ctx, r.Client, resolution.Sources) + if err != nil { + return err + } + // Once a reference has been pinned, a same-name source recreation is not a + // rebind operation. It must be rejected and recreated explicitly by the + // user, rather than silently granting a different backing volume. + if len(ds.Status.MountSources) != 0 && !reflect.DeepEqual(ds.Status.MountSources, bindings) { + return fmt.Errorf("reference mount sources no longer match their recorded identities") + } + if err := mountpolicy.VerifyPVCBinding(ctx, r.Client, ds, resolution.Sources[0], bindings[0]); err != nil { + return err + } + ds.Status.ReadOnly = resolution.ReadOnly + ds.Status.MountSources = bindings + return nil +} + +func (r *DatasetReconciler) setMountPolicyCondition(ds *datasetv1alpha1.Dataset, err error) { + status := metav1.ConditionTrue + reason := "MountPolicyResolved" + message := "" + if err != nil { + status = metav1.ConditionFalse + reason = "MountPolicyDenied" + message = err.Error() + // Keep the stored boolean conservative for older consumers too. New + // consumers must still require the current-generation condition. + ds.Status.ReadOnly = true + } + for i := range ds.Status.Conditions { + condition := &ds.Status.Conditions[i] + if condition.Type != condTypeMountPolicy { + continue + } + if condition.Status != status { + condition.LastTransitionTime = metav1.Now() + } + condition.Status = status + condition.Reason = reason + condition.Message = message + condition.ObservedGeneration = ds.Generation + return + } + ds.Status.Conditions = append(ds.Status.Conditions, metav1.Condition{ + Type: condTypeMountPolicy, + Status: status, + Reason: reason, + Message: message, + ObservedGeneration: ds.Generation, + LastTransitionTime: metav1.Now(), + }) +} + +func referencePVName(ds *datasetv1alpha1.Dataset) string { + uid := string(ds.UID) + if len(uid) > 12 { + uid = uid[:12] + } + if uid == "" { + uid = "pending" + } + return fmt.Sprintf("dataset-%s-%s-%s", ds.Namespace, ds.Name, uid) +} + +func copyStringMap(in map[string]string) map[string]string { + if len(in) == 0 { + return nil + } + out := make(map[string]string, len(in)) + for key, value := range in { + out[key] = value + } + return out +} func (r *DatasetReconciler) reconcilePhase(_ context.Context, ds *datasetv1alpha1.Dataset) error { var phase datasetv1alpha1.DatasetStatusPhase switch ds.Spec.Source.Type { @@ -921,6 +1041,12 @@ func (r *DatasetReconciler) reconcilePhase(_ context.Context, ds *datasetv1alpha ds.Status.Phase = datasetv1alpha1.DatasetStatusPhaseFailed return nil } + if ds.Status.PVCName == "" || !kubeutils.IsConditionReady(ds.Status.Conditions, condTypePVC) || !kubeutils.IsConditionReady(ds.Status.Conditions, condTypeMountPolicy) { + ds.Status.Phase = datasetv1alpha1.DatasetStatusPhasePending + } else { + ds.Status.Phase = datasetv1alpha1.DatasetStatusPhaseReady + } + return nil case datasetv1alpha1.DatasetTypeManual: if _, ok := lo.Find(ds.Status.Conditions, func(c metav1.Condition) bool { return c.Status == metav1.ConditionFalse @@ -952,12 +1078,12 @@ func (r *DatasetReconciler) reconcilePhase(_ context.Context, ds *datasetv1alpha } func (r *DatasetReconciler) getSourceDataset(ctx context.Context, ds *datasetv1alpha1.Dataset) (*datasetv1alpha1.Dataset, error) { - u, err := url.Parse(ds.Spec.Source.URI) + key, err := mountpolicy.ParseReference(ds.Spec.Source.URI) if err != nil { return nil, err } sourceDs := &datasetv1alpha1.Dataset{} - if err := r.Get(ctx, client.ObjectKey{Namespace: u.Host, Name: strings.Trim(u.Path, "/")}, sourceDs); err != nil { + if err := r.Get(ctx, key, sourceDs); err != nil { return nil, fmt.Errorf("fetch source dataset %s error: %v", ds.Spec.Source.URI, err) } return sourceDs, nil @@ -967,36 +1093,25 @@ func (r *DatasetReconciler) validate(ctx context.Context, ds *datasetv1alpha1.Da if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeManual && ds.Spec.Source.URI != "manual://" { return fmt.Errorf("MANUAL dataset source URI must be manual://") } - + if err := mountpolicy.Validate(ds); err != nil { + return err + } if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeReference { - sourceDs, err := r.getSourceDataset(ctx, ds) - if err != nil { + if _, err := mountpolicy.Resolve(ctx, r.Client, ds); err != nil { return err } - if !sourceDs.Spec.Share { - return fmt.Errorf("source dataset %s is not shared", ds.Spec.Source.URI) - } - if sourceDs.Spec.ShareToNamespaceSelector != nil { - // 获取当前 Dataset 所在的 Namespace - currNS := &corev1.Namespace{} - if err := r.Get(ctx, client.ObjectKey{Name: ds.Namespace}, currNS); err != nil { - return fmt.Errorf("fetch current namespace %s error: %v", ds.Namespace, err) - } - s, err := metav1.LabelSelectorAsSelector(sourceDs.Spec.ShareToNamespaceSelector) - if err != nil { - return fmt.Errorf("parse share to namespace selector error: %v", err) - } - if !s.Matches(labels.Set(currNS.Labels)) { - return fmt.Errorf("source dataset %s is not shared to current namespace", ds.Spec.Source.URI) - } - } } - if ds.Spec.VolumeClaimRef != nil && !reflect.DeepEqual(ds.Spec.VolumeClaimTemplate, corev1.PersistentVolumeClaim{}) { return fmt.Errorf("volumeClaimRef and volumeClaimTemplate cannot be both set") } - if ds.Spec.VolumeClaimRef != nil { + protected, err := mountpolicy.ProtectedPVC(ctx, r.Client, ds.Namespace, ds.Spec.VolumeClaimRef.Name) + if err != nil { + return err + } + if protected { + return fmt.Errorf("pvc %s/%s is managed by a REFERENCE dataset; use the dataset reference instead", ds.Namespace, ds.Spec.VolumeClaimRef.Name) + } if ds.Spec.VolumeClaimRef.SubPath != "" { if strings.HasPrefix(ds.Spec.VolumeClaimRef.SubPath, "/") { return fmt.Errorf("subPath should not start with '/', got: %s", ds.Spec.VolumeClaimRef.SubPath) @@ -1006,10 +1121,21 @@ func (r *DatasetReconciler) validate(ctx context.Context, ds *datasetv1alpha1.Da } } } - + if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypePVC { + u, err := url.Parse(ds.Spec.Source.URI) + if err != nil || u.Host == "" { + return fmt.Errorf("invalid PVC dataset uri %q", ds.Spec.Source.URI) + } + protected, err := mountpolicy.ProtectedPVC(ctx, r.Client, ds.Namespace, u.Host) + if err != nil { + return err + } + if protected { + return fmt.Errorf("pvc %s/%s is managed by a REFERENCE dataset; use the dataset reference instead", ds.Namespace, u.Host) + } + } return nil } - func (r *DatasetReconciler) reconcileCascadingDeletion(ctx context.Context, ds *datasetv1alpha1.Dataset) error { // Only perform cascading deletion if enabled in configuration if !config.IsCascadingDeletionEnabled() { @@ -1066,7 +1192,7 @@ func (r *DatasetReconciler) findReferencingDatasets(ctx context.Context, sourceD func (r *DatasetReconciler) cleanupRetainedPV(ctx context.Context, ds *datasetv1alpha1.Dataset) error { // For reference datasets, look for PVs that were created for this dataset // They follow the naming pattern: dataset-{namespace}-{name}-{uid-prefix} - pvName := fmt.Sprintf("dataset-%s-%s-%s", ds.Namespace, ds.Name, ds.UID[:12]) + pvName := referencePVName(ds) pv := &corev1.PersistentVolume{} err := r.Get(ctx, client.ObjectKey{Name: pvName}, pv) @@ -1092,9 +1218,87 @@ func (r *DatasetReconciler) cleanupRetainedPV(ctx context.Context, ds *datasetv1 return nil } +func (r *DatasetReconciler) enqueueReferenceDatasets(ctx context.Context, object client.Object) []reconcile.Request { + list := &datasetv1alpha1.DatasetList{} + if err := r.List(ctx, list); err != nil { + log.Errorf("list reference datasets for policy requeue: %v", err) + return nil + } + + requests := make([]reconcile.Request, 0) + seen := make(map[client.ObjectKey]struct{}) + appendRequest := func(key client.ObjectKey) { + if _, ok := seen[key]; ok { + return + } + seen[key] = struct{}{} + requests = append(requests, reconcile.Request{NamespacedName: key}) + } + + switch changed := object.(type) { + case *corev1.Namespace: + // Namespace labels affect only references whose final target is this namespace. + for i := range list.Items { + ds := &list.Items[i] + if ds.Namespace == changed.Name && ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeReference && !kubeutils.IsDeleted(ds) { + appendRequest(client.ObjectKeyFromObject(ds)) + } + } + return requests + case *datasetv1alpha1.Dataset: + // Walk the reverse reference graph so source changes reach direct and + // transitive dependents without requeueing unrelated references. + reverse := make(map[client.ObjectKey][]client.ObjectKey) + for i := range list.Items { + ds := &list.Items[i] + if ds.Spec.Source.Type != datasetv1alpha1.DatasetTypeReference || kubeutils.IsDeleted(ds) { + continue + } + source, err := mountpolicy.ParseReference(ds.Spec.Source.URI) + if err != nil { + continue + } + reverse[source] = append(reverse[source], client.ObjectKeyFromObject(ds)) + } + + queue := []client.ObjectKey{client.ObjectKeyFromObject(changed)} + visited := make(map[client.ObjectKey]struct{}) + for len(queue) > 0 { + key := queue[0] + queue = queue[1:] + if _, ok := visited[key]; ok { + continue + } + visited[key] = struct{}{} + for _, dependent := range reverse[key] { + appendRequest(dependent) + queue = append(queue, dependent) + } + } + return requests + default: + return nil + } +} + +func dependencyDatasetChanged(update event.UpdateEvent) bool { + oldDataset, oldOK := update.ObjectOld.(*datasetv1alpha1.Dataset) + newDataset, newOK := update.ObjectNew.(*datasetv1alpha1.Dataset) + if !oldOK || !newOK { + return true + } + // Generation tracks policy and source-spec changes. PVCName is status data + // that determines whether a waiting reference can construct its clone. + return oldDataset.Generation != newDataset.Generation || oldDataset.Status.PVCName != newDataset.Status.PVCName + +} + // SetupWithManager sets up the controller with the Manager. func (r *DatasetReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). For(&datasetv1alpha1.Dataset{}). + // Source policy/storage and target namespace label changes invalidate references. + Watches(&datasetv1alpha1.Dataset{}, handler.EnqueueRequestsFromMapFunc(r.enqueueReferenceDatasets), builder.WithPredicates(predicate.Funcs{UpdateFunc: dependencyDatasetChanged})). + Watches(&corev1.Namespace{}, handler.EnqueueRequestsFromMapFunc(r.enqueueReferenceDatasets), builder.WithPredicates(predicate.LabelChangedPredicate{})). Complete(r) } diff --git a/internal/controller/dataset/dataset_controller_test.go b/internal/controller/dataset/dataset_controller_test.go index f9d805a..57f6e4e 100644 --- a/internal/controller/dataset/dataset_controller_test.go +++ b/internal/controller/dataset/dataset_controller_test.go @@ -30,10 +30,14 @@ import ( "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/reconcile" datasetv1alpha1 "github.com/BaizeAI/dataset/api/dataset/v1alpha1" "github.com/BaizeAI/dataset/config" "github.com/BaizeAI/dataset/internal/pkg/constants" + "github.com/BaizeAI/dataset/pkg/kubeutils" + "github.com/BaizeAI/dataset/pkg/mountpolicy" ) func TestDatasetReconciler_findReferencingDatasets(t *testing.T) { @@ -126,6 +130,37 @@ func TestDatasetReconciler_findReferencingDatasets(t *testing.T) { assert.False(t, foundNames["source-dataset"]) } +func TestDatasetReconciler_enqueueReferenceDatasetsScopesDependencies(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, datasetv1alpha1.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + + source := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "source", Namespace: "origin"}} + middle := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "middle", Namespace: "middle"}, Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://origin/source"}}} + leaf := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "leaf", Namespace: "target"}, Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://middle/middle"}}} + unrelated := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "unrelated", Namespace: "other"}, Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://other/root"}}} + reconciler := &DatasetReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(source, middle, leaf, unrelated).Build(), Scheme: scheme} + + requests := reconciler.enqueueReferenceDatasets(context.Background(), source) + require.ElementsMatch(t, []client.ObjectKey{{Namespace: "middle", Name: "middle"}, {Namespace: "target", Name: "leaf"}}, requestKeys(requests)) + + requests = reconciler.enqueueReferenceDatasets(context.Background(), &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "target"}}) + require.Equal(t, []client.ObjectKey{{Namespace: "target", Name: "leaf"}}, requestKeys(requests)) + + oldDataset := source.DeepCopy() + newDataset := source.DeepCopy() + require.False(t, dependencyDatasetChanged(event.UpdateEvent{ObjectOld: oldDataset, ObjectNew: newDataset})) + newDataset.Status.PVCName = "source-pvc" + require.True(t, dependencyDatasetChanged(event.UpdateEvent{ObjectOld: oldDataset, ObjectNew: newDataset})) +} + +func requestKeys(requests []reconcile.Request) []client.ObjectKey { + keys := make([]client.ObjectKey, 0, len(requests)) + for _, request := range requests { + keys = append(keys, request.NamespacedName) + } + return keys +} func TestDatasetReconciler_reconcileCascadingDeletion_Disabled(t *testing.T) { scheme := runtime.NewScheme() require.NoError(t, datasetv1alpha1.AddToScheme(scheme)) @@ -725,3 +760,76 @@ func TestDatasetReconciler_validateManualURI(t *testing.T) { }) } } + +func TestDatasetReconciler_reconcileMountPolicyStoresEffectivePermissions(t *testing.T) { + ctx := context.Background() + scheme := runtime.NewScheme() + require.NoError(t, datasetv1alpha1.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + + rootUID := types.UID("root-uid") + rootPVCUID := types.UID("root-pvc-uid") + rootPVUID := types.UID("root-pv-uid") + refUID := types.UID("ref-uid") + refPVCUID := types.UID("ref-pvc-uid") + refPVUID := types.UID("ref-pv-uid") + root := &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: "root", Namespace: "root", UID: rootUID}, + Spec: datasetv1alpha1.DatasetSpec{ + Share: true, + ShareAccess: &datasetv1alpha1.ShareAccess{Rules: []datasetv1alpha1.ShareAccessRule{{ + NamespaceSelector: metav1.LabelSelector{MatchLabels: map[string]string{"workspace": "one"}}, + AccessMode: datasetv1alpha1.AccessModeReadWrite, + }}}, + Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeManual, URI: "manual://"}, + }, + Status: datasetv1alpha1.DatasetStatus{PVCName: "root-pvc"}, + } + rootPVC := &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "root-pvc", Namespace: "root", UID: rootPVCUID}, + Spec: corev1.PersistentVolumeClaimSpec{VolumeName: "root-pv"}, + Status: corev1.PersistentVolumeClaimStatus{Phase: corev1.ClaimBound}, + } + rootPV := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "root-pv", UID: rootPVUID}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "root", Name: "root-pvc", UID: rootPVCUID}, + PersistentVolumeSource: corev1.PersistentVolumeSource{NFS: &corev1.NFSVolumeSource{Server: "nfs", Path: "/dataset"}}, + }, + } + ref := &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: "ref", Namespace: "target", UID: refUID, Generation: 3}, + Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{ + Type: datasetv1alpha1.DatasetTypeReference, + URI: "dataset://root/root", + }}, + Status: datasetv1alpha1.DatasetStatus{PVCName: "ref-pvc"}, + } + owner := *metav1.NewControllerRef(ref, datasetv1alpha1.GroupVersion.WithKind("Dataset")) + refPVC := &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "ref-pvc", Namespace: "target", UID: refPVCUID, OwnerReferences: []metav1.OwnerReference{owner}}, + Spec: corev1.PersistentVolumeClaimSpec{VolumeName: "ref-pv"}, + Status: corev1.PersistentVolumeClaimStatus{Phase: corev1.ClaimBound}, + } + refPV := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "ref-pv", UID: refPVUID, OwnerReferences: []metav1.OwnerReference{owner}, Annotations: map[string]string{ + mountpolicy.SourceDatasetUIDAnnotation: string(rootUID), + mountpolicy.SourcePVCUIDAnnotation: string(rootPVCUID), + mountpolicy.SourcePVUIDAnnotation: string(rootPVUID), + }}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "target", Name: "ref-pvc", UID: refPVCUID}, + PersistentVolumeSource: corev1.PersistentVolumeSource{NFS: &corev1.NFSVolumeSource{Server: "nfs", Path: "/dataset"}}, + }, + } + workspace := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "target", Labels: map[string]string{"workspace": "one"}}} + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(root, rootPVC, rootPV, refPVC, refPV, workspace).Build() + reconciler := &DatasetReconciler{Client: fakeClient, Scheme: scheme} + + require.NoError(t, reconciler.reconcileMountPolicy(ctx, ref)) + reconciler.setMountPolicyCondition(ref, nil) + require.False(t, ref.Status.ReadOnly) + require.Equal(t, []datasetv1alpha1.MountSource{{Namespace: "root", Name: "root", UID: string(rootUID), PVCName: "root-pvc", PVCUID: string(rootPVCUID), PVName: "root-pv", PVUID: string(rootPVUID)}}, ref.Status.MountSources) + require.True(t, kubeutils.IsConditionReady(ref.Status.Conditions, condTypeMountPolicy)) + require.Equal(t, int64(3), ref.Status.Conditions[0].ObservedGeneration) +} diff --git a/internal/controller/dataset/mount_policy_additional_test.go b/internal/controller/dataset/mount_policy_additional_test.go new file mode 100644 index 0000000..978f4fd --- /dev/null +++ b/internal/controller/dataset/mount_policy_additional_test.go @@ -0,0 +1,94 @@ +package dataset + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + datasetv1alpha1 "github.com/BaizeAI/dataset/api/dataset/v1alpha1" + "github.com/BaizeAI/dataset/pkg/mountpolicy" +) + +func TestReconcileMountPolicyStoresFullSourceChain(t *testing.T) { + ctx := context.Background() + scheme := runtime.NewScheme() + require.NoError(t, datasetv1alpha1.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + + a := referenceTestSource("a", "a", "a-uid", "a-pvc", "a-pvc-uid", "a-pv", "a-pv-uid", nil, nil) + a.Spec.Share = true + b := referenceTestSource("b", "b", "b-uid", "b-pvc", "b-pvc-uid", "b-pv", "b-pv-uid", a, &datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://a/a"}) + c := referenceTestSource("c", "target", "c-uid", "c-pvc", "c-pvc-uid", "c-pv", "c-pv-uid", b, &datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://b/b"}) + c.Spec.Share = false + objects := append(referenceTestStorage(a, nil), referenceTestStorage(b, a)...) + objects = append(objects, referenceTestStorage(c, b)...) + objects = append(objects, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "target", Labels: map[string]string{"workspace": "one"}}}) + reconciler := &DatasetReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build(), Scheme: scheme} + + require.NoError(t, reconciler.reconcileMountPolicy(ctx, c)) + reconciler.setMountPolicyCondition(c, nil) + require.Equal(t, []datasetv1alpha1.MountSource{ + {Namespace: "b", Name: "b", UID: "b-uid", PVCName: "b-pvc", PVCUID: "b-pvc-uid", PVName: "b-pv", PVUID: "b-pv-uid"}, + {Namespace: "a", Name: "a", UID: "a-uid", PVCName: "a-pvc", PVCUID: "a-pvc-uid", PVName: "a-pv", PVUID: "a-pv-uid"}, + }, c.Status.MountSources) + require.False(t, c.Status.ReadOnly) +} + +func TestReferenceFailedRequeuesWithoutWatchEvent(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, datasetv1alpha1.AddToScheme(scheme)) + ref := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "waiting", Namespace: "target"}, Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://origin/missing"}}} + reconciler := &DatasetReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(ref).WithObjects(ref).Build(), Scheme: scheme} + + result, err := reconciler.Reconcile(context.Background(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(ref)}) + require.NoError(t, err) + require.Equal(t, 30*time.Second, result.RequeueAfter) +} + +func TestReferencePVNameAcceptsShortUID(t *testing.T) { + require.Equal(t, "dataset-team-data-short", referencePVName(&datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Namespace: "team", Name: "data", UID: types.UID("short")}})) + require.Equal(t, "dataset-team-data-pending", referencePVName(&datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Namespace: "team", Name: "data"}})) +} + +func referenceTestSource(name, namespace, uid, pvcName, pvcUID, pvName, pvUID string, parent *datasetv1alpha1.Dataset, source *datasetv1alpha1.DatasetSource) *datasetv1alpha1.Dataset { + dsSource := datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeManual, URI: "manual://"} + if source != nil { + dsSource = *source + } + return &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace, UID: types.UID(uid)}, + Spec: datasetv1alpha1.DatasetSpec{ + Share: parent != nil, + ShareAccess: &datasetv1alpha1.ShareAccess{Rules: []datasetv1alpha1.ShareAccessRule{{NamespaceSelector: metav1.LabelSelector{MatchLabels: map[string]string{"workspace": "one"}}, AccessMode: datasetv1alpha1.AccessModeReadWrite}}}, + Source: dsSource, + }, + Status: datasetv1alpha1.DatasetStatus{PVCName: pvcName}, + } +} + +func referenceTestStorage(ds, parent *datasetv1alpha1.Dataset) []client.Object { + pvcUID := types.UID(ds.Name + "-pvc-uid") + pvUID := types.UID(ds.Name + "-pv-uid") + pvc := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: ds.Status.PVCName, Namespace: ds.Namespace, UID: pvcUID}, Spec: corev1.PersistentVolumeClaimSpec{VolumeName: ds.Name + "-pv"}, Status: corev1.PersistentVolumeClaimStatus{Phase: corev1.ClaimBound}} + pv := &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: ds.Name + "-pv", UID: pvUID}, Spec: corev1.PersistentVolumeSpec{ClaimRef: &corev1.ObjectReference{Namespace: ds.Namespace, Name: pvc.Name, UID: pvc.UID}, PersistentVolumeSource: corev1.PersistentVolumeSource{NFS: &corev1.NFSVolumeSource{Server: "nfs", Path: "/data"}}}} + if parent != nil { + owner := *metav1.NewControllerRef(ds, datasetv1alpha1.GroupVersion.WithKind("Dataset")) + pvc.OwnerReferences = []metav1.OwnerReference{owner} + pv.OwnerReferences = []metav1.OwnerReference{owner} + pv.Annotations = map[string]string{ + mountpolicy.SourceDatasetUIDAnnotation: string(parent.UID), + mountpolicy.SourcePVCUIDAnnotation: parent.Name + "-pvc-uid", + mountpolicy.SourcePVUIDAnnotation: parent.Name + "-pv-uid", + } + } + return []client.Object{ds, pvc, pv} +} diff --git a/pkg/datasource/huggingface/fake/hub.go b/pkg/datasource/huggingface/fake/hub.go index 4ec1972..24410bc 100644 --- a/pkg/datasource/huggingface/fake/hub.go +++ b/pkg/datasource/huggingface/fake/hub.go @@ -4,7 +4,7 @@ package fake import ( "context" "sync" - + "github.com/BaizeAI/dataset/pkg/datasource/huggingface" ) diff --git a/pkg/datasource/modelscope/fake/hub.go b/pkg/datasource/modelscope/fake/hub.go index f6ed885..01bb6c7 100644 --- a/pkg/datasource/modelscope/fake/hub.go +++ b/pkg/datasource/modelscope/fake/hub.go @@ -4,7 +4,7 @@ package fake import ( "context" "sync" - + "github.com/BaizeAI/dataset/pkg/datasource/modelscope" ) diff --git a/pkg/mountpolicy/policy.go b/pkg/mountpolicy/policy.go new file mode 100644 index 0000000..9639f40 --- /dev/null +++ b/pkg/mountpolicy/policy.go @@ -0,0 +1,422 @@ +// Package mountpolicy resolves and verifies Dataset REFERENCE mount permissions. +// +// It is intentionally independent from the Dataset controller so every mount +// consumer can make the same fail-closed decision immediately before creating a +// Pod. +package mountpolicy + +import ( + "context" + "fmt" + "net/url" + "reflect" + "strings" + + datasetv1alpha1 "github.com/BaizeAI/dataset/api/dataset/v1alpha1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const ( + // MaxReferenceDepth bounds both the work performed by a request and the + // amount of source state a Dataset status can retain. + MaxReferenceDepth = 32 + // MountPolicyCondition is the authoritative condition for status.ReadOnly. + MountPolicyCondition = "MountPolicy" + + SourceDatasetUIDAnnotation = "dataset.baizeai.io/mount-source-dataset-uid" + SourcePVCUIDAnnotation = "dataset.baizeai.io/mount-source-pvc-uid" + SourcePVUIDAnnotation = "dataset.baizeai.io/mount-source-pv-uid" +) + +// GrantResult is the result of granting one source Dataset to one target +// namespace. It is not a Kubernetes volume access mode. +type GrantResult string + +const ( + Denied GrantResult = "Denied" + ReadOnly GrantResult = "ReadOnly" + ReadWrite GrantResult = "ReadWrite" +) + +// Resolution is the direct-to-root source chain for one REFERENCE Dataset. +type Resolution struct { + Sources []*datasetv1alpha1.Dataset + ReadOnly bool +} + +// Validate verifies policy data defensively. CRD CEL performs the same checks +// for normal API writes, but controllers must not trust objects that bypassed +// schema validation (for example, a fake client, an old API server, or an +// imported object). +func Validate(ds *datasetv1alpha1.Dataset) error { + if ds == nil { + return fmt.Errorf("dataset is required") + } + if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeReference && ds.Spec.VolumeClaimRef != nil { + return fmt.Errorf("REFERENCE dataset cannot set volumeClaimRef") + } + if ds.Spec.ShareToNamespaceSelector != nil && !emptySelector(ds.Spec.ShareToNamespaceSelector) { + if _, err := metav1.LabelSelectorAsSelector(ds.Spec.ShareToNamespaceSelector); err != nil { + return fmt.Errorf("shareToNamespaceSelector is invalid: %w", err) + } + } + policy := ds.Spec.ShareAccess + if policy == nil { + return nil + } + if len(policy.Rules) == 0 { + return fmt.Errorf("shareAccess.rules must contain at least one rule when shareAccess is configured") + } + for i, rule := range policy.Rules { + if rule.AccessMode != datasetv1alpha1.AccessModeReadOnly && rule.AccessMode != datasetv1alpha1.AccessModeReadWrite { + return fmt.Errorf("shareAccess.rules[%d].accessMode %q is invalid", i, rule.AccessMode) + } + if emptySelector(&rule.NamespaceSelector) { + return fmt.Errorf("shareAccess.rules[%d].namespaceSelector must not be empty", i) + } + if _, err := metav1.LabelSelectorAsSelector(&rule.NamespaceSelector); err != nil { + return fmt.Errorf("shareAccess.rules[%d].namespaceSelector is invalid: %w", i, err) + } + } + return nil +} + +// Grant computes the access that source grants to targetNamespace. Namespace +// labels are always loaded from the API reader; callers never supply a +// workspace ID or a label map, which prevents a request from forging identity. +func Grant(ctx context.Context, reader client.Reader, source *datasetv1alpha1.Dataset, targetNamespace string) (GrantResult, error) { + if source == nil { + return Denied, fmt.Errorf("source dataset is required") + } + if targetNamespace == "" { + return Denied, fmt.Errorf("target namespace is required") + } + if source.DeletionTimestamp != nil || !source.Spec.Share { + return Denied, nil + } + if err := Validate(source); err != nil { + return Denied, err + } + + ns := &corev1.Namespace{} + if err := reader.Get(ctx, client.ObjectKey{Name: targetNamespace}, ns); err != nil { + return Denied, fmt.Errorf("get target namespace %q: %w", targetNamespace, err) + } + if source.Spec.ShareToNamespaceSelector != nil && !emptySelector(source.Spec.ShareToNamespaceSelector) { + selector, err := metav1.LabelSelectorAsSelector(source.Spec.ShareToNamespaceSelector) + if err != nil { + return Denied, fmt.Errorf("parse shareToNamespaceSelector: %w", err) + } + if !selector.Matches(labels.Set(ns.Labels)) { + return Denied, nil + } + } + + // A nil policy is the legacy sharing contract: shared references are + // readable, never writable. An explicit policy has no fallback rule. + if source.Spec.ShareAccess == nil { + return ReadOnly, nil + } + + grant := Denied + for i, rule := range source.Spec.ShareAccess.Rules { + selector, err := metav1.LabelSelectorAsSelector(&rule.NamespaceSelector) + if err != nil { + return Denied, fmt.Errorf("parse shareAccess.rules[%d].namespaceSelector: %w", i, err) + } + if !selector.Matches(labels.Set(ns.Labels)) { + continue + } + // ReadOnly wins over every ReadWrite match. + if rule.AccessMode == datasetv1alpha1.AccessModeReadOnly { + return ReadOnly, nil + } + grant = ReadWrite + } + return grant, nil +} + +// Resolve re-resolves every reference edge against the final Dataset namespace. +// It does not use upstream status.ReadOnly: status is a cached outcome and +// cannot grant access after a source policy changes. +func Resolve(ctx context.Context, reader client.Reader, ds *datasetv1alpha1.Dataset) (*Resolution, error) { + if ds == nil || ds.Spec.Source.Type != datasetv1alpha1.DatasetTypeReference { + return nil, fmt.Errorf("a REFERENCE dataset is required") + } + if err := Validate(ds); err != nil { + return nil, err + } + + seen := map[string]struct{}{datasetKey(ds.Namespace, ds.Name): {}} + current := ds + result := &Resolution{} + for depth := 0; ; depth++ { + if depth >= MaxReferenceDepth { + return nil, fmt.Errorf("reference chain exceeds maximum depth %d", MaxReferenceDepth) + } + ref, err := ParseReference(current.Spec.Source.URI) + if err != nil { + return nil, err + } + key := datasetKey(ref.Namespace, ref.Name) + if _, ok := seen[key]; ok { + return nil, fmt.Errorf("reference cycle detected at %s/%s", ref.Namespace, ref.Name) + } + seen[key] = struct{}{} + + source := &datasetv1alpha1.Dataset{} + if err := reader.Get(ctx, ref, source); err != nil { + return nil, fmt.Errorf("get source dataset %s/%s: %w", ref.Namespace, ref.Name, err) + } + if source.DeletionTimestamp != nil { + return nil, fmt.Errorf("source dataset %s/%s is deleting", source.Namespace, source.Name) + } + grant, err := Grant(ctx, reader, source, ds.Namespace) + if err != nil { + return nil, err + } + if grant == Denied { + return nil, fmt.Errorf("source dataset %s/%s is not shared to namespace %s", source.Namespace, source.Name, ds.Namespace) + } + result.Sources = append(result.Sources, source) + if grant == ReadOnly { + result.ReadOnly = true + } + if source.Spec.Source.Type != datasetv1alpha1.DatasetTypeReference { + return result, nil + } + if source.Spec.VolumeClaimRef != nil { + return nil, fmt.Errorf("source REFERENCE dataset %s/%s sets volumeClaimRef", source.Namespace, source.Name) + } + current = source + } +} + +// Bindings returns the status representation of the supplied source chain and +// verifies that every Dataset has an actual PVC and PV. It intentionally reads +// live objects rather than trusting an old status record. +func Bindings(ctx context.Context, reader client.Reader, sources []*datasetv1alpha1.Dataset) ([]datasetv1alpha1.MountSource, error) { + if len(sources) == 0 { + return nil, fmt.Errorf("reference source chain is empty") + } + bindings := make([]datasetv1alpha1.MountSource, 0, len(sources)) + for _, source := range sources { + if source == nil || source.Status.PVCName == "" { + return nil, fmt.Errorf("source dataset has no pvc") + } + pvc := &corev1.PersistentVolumeClaim{} + if err := reader.Get(ctx, client.ObjectKey{Namespace: source.Namespace, Name: source.Status.PVCName}, pvc); err != nil { + return nil, fmt.Errorf("get source pvc %s/%s: %w", source.Namespace, source.Status.PVCName, err) + } + if pvc.Status.Phase != corev1.ClaimBound || pvc.Spec.VolumeName == "" { + return nil, fmt.Errorf("source pvc %s/%s is not bound", pvc.Namespace, pvc.Name) + } + pv := &corev1.PersistentVolume{} + if err := reader.Get(ctx, client.ObjectKey{Name: pvc.Spec.VolumeName}, pv); err != nil { + return nil, fmt.Errorf("get source pv %s: %w", pvc.Spec.VolumeName, err) + } + if pv.Spec.ClaimRef == nil || pv.Spec.ClaimRef.Namespace != pvc.Namespace || pv.Spec.ClaimRef.Name != pvc.Name || pv.Spec.ClaimRef.UID != pvc.UID { + return nil, fmt.Errorf("source pv %s is not bound to pvc %s/%s", pv.Name, pvc.Namespace, pvc.Name) + } + bindings = append(bindings, datasetv1alpha1.MountSource{ + Namespace: source.Namespace, + Name: source.Name, + UID: string(source.UID), + PVCName: pvc.Name, + PVCUID: string(pvc.UID), + PVName: pv.Name, + PVUID: string(pv.UID), + }) + } + return bindings, nil +} + +// Verify confirms that a reference Dataset is still mountable. It is the only +// method a pod-producing consumer should use to decide whether ReadOnly=false +// permits a writable volume. +func Verify(ctx context.Context, reader client.Reader, ds *datasetv1alpha1.Dataset) (*Resolution, error) { + if ds == nil || ds.Spec.Source.Type != datasetv1alpha1.DatasetTypeReference { + return nil, fmt.Errorf("a REFERENCE dataset is required") + } + if !mountPolicyReady(ds) { + return nil, fmt.Errorf("MountPolicy is not ready for current generation") + } + resolution, err := Resolve(ctx, reader, ds) + if err != nil { + return nil, err + } + bindings, err := Bindings(ctx, reader, resolution.Sources) + if err != nil { + return nil, err + } + if !reflect.DeepEqual(ds.Status.MountSources, bindings) { + return nil, fmt.Errorf("mount source bindings no longer match status") + } + if ds.Status.ReadOnly != resolution.ReadOnly { + return nil, fmt.Errorf("mount access mode no longer matches source policy") + } + if err := VerifyPVCBinding(ctx, reader, ds, resolution.Sources[0], bindings[0]); err != nil { + return nil, err + } + for i, source := range resolution.Sources[:len(resolution.Sources)-1] { + if source.Spec.Source.Type != datasetv1alpha1.DatasetTypeReference { + continue + } + if err := verifyReferencePV(ctx, reader, source, bindings[i], bindings[i+1]); err != nil { + return nil, err + } + } + return resolution, nil +} + +// VerifyPVCBinding validates the current reference Dataset's own PVC/PV. The +// source Dataset/PV are passed explicitly so this is also usable by the +// controller before it writes the MountPolicy=True condition. +func VerifyPVCBinding(ctx context.Context, reader client.Reader, ds *datasetv1alpha1.Dataset, source *datasetv1alpha1.Dataset, sourceBinding datasetv1alpha1.MountSource) error { + if ds == nil || source == nil || ds.Status.PVCName == "" { + return fmt.Errorf("reference dataset or its pvc is missing") + } + pvc := &corev1.PersistentVolumeClaim{} + if err := reader.Get(ctx, client.ObjectKey{Namespace: ds.Namespace, Name: ds.Status.PVCName}, pvc); err != nil { + return fmt.Errorf("get reference pvc %s/%s: %w", ds.Namespace, ds.Status.PVCName, err) + } + if !ownedBy(pvc.OwnerReferences, ds) { + return fmt.Errorf("reference pvc %s/%s is not owned by dataset", pvc.Namespace, pvc.Name) + } + if pvc.Status.Phase != corev1.ClaimBound || pvc.Spec.VolumeName == "" { + return fmt.Errorf("reference pvc %s/%s is not bound", pvc.Namespace, pvc.Name) + } + pv := &corev1.PersistentVolume{} + if err := reader.Get(ctx, client.ObjectKey{Name: pvc.Spec.VolumeName}, pv); err != nil { + return fmt.Errorf("get reference pv %s: %w", pvc.Spec.VolumeName, err) + } + if !ownedBy(pv.OwnerReferences, ds) { + return fmt.Errorf("reference pv %s is not owned by dataset", pv.Name) + } + if pv.Spec.ClaimRef == nil || pv.Spec.ClaimRef.Namespace != pvc.Namespace || pv.Spec.ClaimRef.Name != pvc.Name || pv.Spec.ClaimRef.UID != pvc.UID { + return fmt.Errorf("reference pv %s is not bound to its pvc", pv.Name) + } + sourcePV := &corev1.PersistentVolume{} + if err := reader.Get(ctx, client.ObjectKey{Name: sourceBinding.PVName}, sourcePV); err != nil { + return fmt.Errorf("get bound source pv %s: %w", sourceBinding.PVName, err) + } + if string(sourcePV.UID) != sourceBinding.PVUID { + return fmt.Errorf("bound source pv %s no longer matches binding", sourceBinding.PVName) + } + return verifyClonePV(pv, source, sourceBinding, sourcePV) +} + +// ProtectedPVC reports whether a PVC is the controller-managed PVC of any +// REFERENCE Dataset. Callers must treat an error as a denial, preventing an +// ordinary PVC Dataset from becoming an alias that bypasses mount policy. +func ProtectedPVC(ctx context.Context, reader client.Reader, namespace, pvcName string) (bool, error) { + if namespace == "" || pvcName == "" { + return false, fmt.Errorf("namespace and pvc name are required") + } + list := &datasetv1alpha1.DatasetList{} + if err := reader.List(ctx, list); err != nil { + return false, fmt.Errorf("list datasets for protected pvc check: %w", err) + } + for i := range list.Items { + ds := &list.Items[i] + if ds.Namespace != namespace || ds.Spec.Source.Type != datasetv1alpha1.DatasetTypeReference { + continue + } + if referencePVCName(ds) == pvcName { + return true, nil + } + } + return false, nil +} + +func verifyReferencePV(ctx context.Context, reader client.Reader, ds *datasetv1alpha1.Dataset, binding, parent datasetv1alpha1.MountSource) error { + pvc := &corev1.PersistentVolumeClaim{} + if err := reader.Get(ctx, client.ObjectKey{Namespace: binding.Namespace, Name: binding.PVCName}, pvc); err != nil { + return fmt.Errorf("get reference source pvc %s/%s: %w", binding.Namespace, binding.PVCName, err) + } + if string(pvc.UID) != binding.PVCUID || pvc.Spec.VolumeName != binding.PVName { + return fmt.Errorf("reference source pvc %s/%s no longer matches binding", binding.Namespace, binding.PVCName) + } + pv := &corev1.PersistentVolume{} + if err := reader.Get(ctx, client.ObjectKey{Name: binding.PVName}, pv); err != nil { + return fmt.Errorf("get reference source pv %s: %w", binding.PVName, err) + } + if string(pv.UID) != binding.PVUID || !ownedBy(pv.OwnerReferences, ds) { + return fmt.Errorf("reference source pv %s no longer matches binding", binding.PVName) + } + parentDS := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Namespace: parent.Namespace, Name: parent.Name, UID: types.UID(parent.UID)}} + parentPV := &corev1.PersistentVolume{} + if err := reader.Get(ctx, client.ObjectKey{Name: parent.PVName}, parentPV); err != nil { + return fmt.Errorf("get parent pv %s: %w", parent.PVName, err) + } + if string(parentPV.UID) != parent.PVUID { + return fmt.Errorf("parent pv %s no longer matches binding", parent.PVName) + } + return verifyClonePV(pv, parentDS, parent, parentPV) +} + +func verifyClonePV(pv *corev1.PersistentVolume, source *datasetv1alpha1.Dataset, sourceBinding datasetv1alpha1.MountSource, sourcePV *corev1.PersistentVolume) error { + if pv.Annotations == nil || + pv.Annotations[SourceDatasetUIDAnnotation] != string(source.UID) || + pv.Annotations[SourcePVCUIDAnnotation] != sourceBinding.PVCUID || + pv.Annotations[SourcePVUIDAnnotation] != sourceBinding.PVUID { + return fmt.Errorf("reference pv %s source identity does not match", pv.Name) + } + if sourcePV == nil || !reflect.DeepEqual(pv.Spec.PersistentVolumeSource, sourcePV.Spec.PersistentVolumeSource) { + return fmt.Errorf("reference pv %s persistent volume source no longer matches", pv.Name) + } + return nil +} + +func mountPolicyReady(ds *datasetv1alpha1.Dataset) bool { + for _, condition := range ds.Status.Conditions { + if condition.Type == MountPolicyCondition && condition.Status == metav1.ConditionTrue && condition.ObservedGeneration == ds.Generation { + return true + } + } + return false +} + +func emptySelector(selector *metav1.LabelSelector) bool { + return selector == nil || (len(selector.MatchLabels) == 0 && len(selector.MatchExpressions) == 0) +} + +// ParseReference validates a REFERENCE URI and returns its source Dataset key. +// It is shared by the controller and policy resolver so both paths accept the +// same URI grammar. +func ParseReference(uri string) (client.ObjectKey, error) { + u, err := url.Parse(uri) + if err != nil { + return client.ObjectKey{}, fmt.Errorf("parse reference uri %q: %w", uri, err) + } + name := strings.Trim(u.Path, "/") + if u.Scheme != "dataset" || u.Host == "" || name == "" || strings.Contains(name, "/") || u.RawQuery != "" || u.Fragment != "" { + return client.ObjectKey{}, fmt.Errorf("invalid reference uri %q, expected dataset:///", uri) + } + return client.ObjectKey{Namespace: u.Host, Name: name}, nil +} + +func datasetKey(namespace, name string) string { return namespace + "/" + name } + +func referencePVCName(ds *datasetv1alpha1.Dataset) string { + if ds.Status.PVCName != "" { + return ds.Status.PVCName + } + if ds.Spec.VolumeClaimTemplate.Name != "" { + return ds.Spec.VolumeClaimTemplate.Name + } + return ds.Name +} + +func ownedBy(refs []metav1.OwnerReference, ds *datasetv1alpha1.Dataset) bool { + for _, ref := range refs { + if ref.APIVersion == datasetv1alpha1.GroupVersion.String() && ref.Kind == "Dataset" && ref.Name == ds.Name && ref.UID == ds.UID { + return true + } + } + return false +} diff --git a/pkg/mountpolicy/policy_identity_test.go b/pkg/mountpolicy/policy_identity_test.go new file mode 100644 index 0000000..891426f --- /dev/null +++ b/pkg/mountpolicy/policy_identity_test.go @@ -0,0 +1,69 @@ +package mountpolicy + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + datasetv1alpha1 "github.com/BaizeAI/dataset/api/dataset/v1alpha1" +) + +func TestVerifyRejectsRecreatedSourceIdentities(t *testing.T) { + ctx := context.Background() + rootUID, rootPVCUID, rootPVUID := types.UID("root"), types.UID("root-pvc"), types.UID("root-pv") + leafUID, leafPVCUID, leafPVUID := types.UID("leaf"), types.UID("leaf-pvc"), types.UID("leaf-pv") + target := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "target", Labels: map[string]string{"workspace": "one"}}} + root := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "root", Namespace: "root", UID: rootUID}, Spec: datasetv1alpha1.DatasetSpec{Share: true, ShareAccess: policy(datasetv1alpha1.AccessModeReadWrite, "one"), Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeManual, URI: "manual://"}}, Status: datasetv1alpha1.DatasetStatus{PVCName: "root-pvc"}} + rootPVC := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "root-pvc", Namespace: "root", UID: rootPVCUID}, Spec: corev1.PersistentVolumeClaimSpec{VolumeName: "root-pv"}, Status: corev1.PersistentVolumeClaimStatus{Phase: corev1.ClaimBound}} + rootPV := &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: "root-pv", UID: rootPVUID}, Spec: corev1.PersistentVolumeSpec{ClaimRef: &corev1.ObjectReference{Namespace: "root", Name: "root-pvc", UID: rootPVCUID}, PersistentVolumeSource: corev1.PersistentVolumeSource{NFS: &corev1.NFSVolumeSource{Server: "nfs", Path: "/data"}}}} + leaf := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "leaf", Namespace: "target", UID: leafUID, Generation: 1}, Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://root/root"}}, Status: datasetv1alpha1.DatasetStatus{PVCName: "leaf-pvc", MountSources: []datasetv1alpha1.MountSource{{Namespace: "root", Name: "root", UID: string(rootUID), PVCName: "root-pvc", PVCUID: string(rootPVCUID), PVName: "root-pv", PVUID: string(rootPVUID)}}, Conditions: []metav1.Condition{{Type: MountPolicyCondition, Status: metav1.ConditionTrue, ObservedGeneration: 1}}}} + owner := *metav1.NewControllerRef(leaf, datasetv1alpha1.GroupVersion.WithKind("Dataset")) + leafPVC := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "leaf-pvc", Namespace: "target", UID: leafPVCUID, OwnerReferences: []metav1.OwnerReference{owner}}, Spec: corev1.PersistentVolumeClaimSpec{VolumeName: "leaf-pv"}, Status: corev1.PersistentVolumeClaimStatus{Phase: corev1.ClaimBound}} + leafPV := &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: "leaf-pv", UID: leafPVUID, OwnerReferences: []metav1.OwnerReference{owner}, Annotations: map[string]string{SourceDatasetUIDAnnotation: string(rootUID), SourcePVCUIDAnnotation: string(rootPVCUID), SourcePVUIDAnnotation: string(rootPVUID)}}, Spec: corev1.PersistentVolumeSpec{ClaimRef: &corev1.ObjectReference{Namespace: "target", Name: "leaf-pvc", UID: leafPVCUID}, PersistentVolumeSource: corev1.PersistentVolumeSource{NFS: &corev1.NFSVolumeSource{Server: "nfs", Path: "/data"}}}} + + _, err := Verify(ctx, newReader(t, target, root, rootPVC, rootPV, leafPVC, leafPV), leaf) + require.NoError(t, err) + + tests := []struct { + name string + objects func() []client.Object + }{ + { + name: "dataset UID", + objects: func() []client.Object { + rebuilt := root.DeepCopy() + rebuilt.UID = types.UID("root-rebuilt") + return []client.Object{target, rebuilt, rootPVC, rootPV, leafPVC, leafPV} + }, + }, + { + name: "PVC UID", + objects: func() []client.Object { + rebuiltPVC := rootPVC.DeepCopy() + rebuiltPVC.UID = types.UID("root-pvc-rebuilt") + rebuiltPV := rootPV.DeepCopy() + rebuiltPV.Spec.ClaimRef.UID = rebuiltPVC.UID + return []client.Object{target, root, rebuiltPVC, rebuiltPV, leafPVC, leafPV} + }, + }, + { + name: "PV UID", + objects: func() []client.Object { + rebuilt := rootPV.DeepCopy() + rebuilt.UID = types.UID("root-pv-rebuilt") + return []client.Object{target, root, rootPVC, rebuilt, leafPVC, leafPV} + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := Verify(ctx, newReader(t, tt.objects()...), leaf) + require.ErrorContains(t, err, "mount source bindings no longer match status") + }) + } +} diff --git a/pkg/mountpolicy/policy_test.go b/pkg/mountpolicy/policy_test.go new file mode 100644 index 0000000..8b8bfc7 --- /dev/null +++ b/pkg/mountpolicy/policy_test.go @@ -0,0 +1,163 @@ +package mountpolicy + +import ( + "context" + "testing" + + datasetv1alpha1 "github.com/BaizeAI/dataset/api/dataset/v1alpha1" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +func policy(mode datasetv1alpha1.AccessMode, value string) *datasetv1alpha1.ShareAccess { + return &datasetv1alpha1.ShareAccess{Rules: []datasetv1alpha1.ShareAccessRule{{ + NamespaceSelector: metav1.LabelSelector{MatchLabels: map[string]string{"workspace": value}}, + AccessMode: mode, + }}} +} + +func newReader(t *testing.T, objects ...client.Object) client.Client { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, datasetv1alpha1.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build() +} + +func TestGrantRulesAndLegacyPolicy(t *testing.T) { + ctx := context.Background() + source := &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: "source", Namespace: "source"}, + Spec: datasetv1alpha1.DatasetSpec{Share: true, ShareAccess: &datasetv1alpha1.ShareAccess{Rules: []datasetv1alpha1.ShareAccessRule{ + {NamespaceSelector: metav1.LabelSelector{MatchLabels: map[string]string{"workspace": "rw"}}, AccessMode: datasetv1alpha1.AccessModeReadWrite}, + {NamespaceSelector: metav1.LabelSelector{MatchLabels: map[string]string{"workspace": "ro"}}, AccessMode: datasetv1alpha1.AccessModeReadOnly}, + // A readonly match wins over a simultaneously matching readwrite rule. + {NamespaceSelector: metav1.LabelSelector{MatchLabels: map[string]string{"workspace": "rw", "tier": "restricted"}}, AccessMode: datasetv1alpha1.AccessModeReadOnly}, + }}}, + } + rw := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "rw", Labels: map[string]string{"workspace": "rw"}}} + restricted := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "restricted", Labels: map[string]string{"workspace": "rw", "tier": "restricted"}}} + ro := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ro", Labels: map[string]string{"workspace": "ro"}}} + denied := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "denied", Labels: map[string]string{"workspace": "other"}}} + reader := newReader(t, rw, restricted, ro, denied) + + result, err := Grant(ctx, reader, source, "rw") + require.NoError(t, err) + require.Equal(t, ReadWrite, result) + result, err = Grant(ctx, reader, source, "restricted") + require.NoError(t, err) + require.Equal(t, ReadOnly, result) + result, err = Grant(ctx, reader, source, "ro") + require.NoError(t, err) + require.Equal(t, ReadOnly, result) + result, err = Grant(ctx, reader, source, "denied") + require.NoError(t, err) + require.Equal(t, Denied, result) + + legacy := source.DeepCopy() + legacy.Spec.ShareAccess = nil + result, err = Grant(ctx, reader, legacy, "denied") + require.NoError(t, err) + require.Equal(t, ReadOnly, result) + + empty := source.DeepCopy() + empty.Spec.ShareAccess = &datasetv1alpha1.ShareAccess{} + _, err = Grant(ctx, reader, empty, "rw") + require.ErrorContains(t, err, "must contain at least one rule") + + preconfigured := source.DeepCopy() + preconfigured.Spec.Share = false + preconfigured.Spec.ShareAccess = &datasetv1alpha1.ShareAccess{} + require.ErrorContains(t, Validate(preconfigured), "must contain at least one rule") +} + +func TestParseReference(t *testing.T) { + key, err := ParseReference("dataset://source/data") + require.NoError(t, err) + require.Equal(t, client.ObjectKey{Namespace: "source", Name: "data"}, key) + + for _, uri := range []string{"https://source/data", "dataset://source/data/child", "dataset://source/data?x=y", "dataset://source/data#fragment"} { + _, err := ParseReference(uri) + require.Error(t, err, uri) + } +} + +func TestResolveRequiresEverySourceAndReadonlyWins(t *testing.T) { + ctx := context.Background() + target := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "target", Labels: map[string]string{"workspace": "one"}}} + root := &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: "root", Namespace: "root", UID: types.UID("root-uid")}, + Spec: datasetv1alpha1.DatasetSpec{Share: true, ShareAccess: policy(datasetv1alpha1.AccessModeReadOnly, "one"), Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeManual, URI: "manual://"}}, + } + middle := &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: "middle", Namespace: "middle", UID: types.UID("middle-uid")}, + Spec: datasetv1alpha1.DatasetSpec{Share: true, ShareAccess: policy(datasetv1alpha1.AccessModeReadWrite, "one"), Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://root/root"}}, + } + leaf := &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: "leaf", Namespace: "target"}, + Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://middle/middle"}}, + } + reader := newReader(t, target, root, middle) + resolution, err := Resolve(ctx, reader, leaf) + require.NoError(t, err) + require.True(t, resolution.ReadOnly) + require.Len(t, resolution.Sources, 2) + require.Equal(t, []string{"middle", "root"}, []string{resolution.Sources[0].Name, resolution.Sources[1].Name}) + + root.Spec.ShareAccess = policy(datasetv1alpha1.AccessModeReadWrite, "other") + reader = newReader(t, target, root, middle) + _, err = Resolve(ctx, reader, leaf) + require.ErrorContains(t, err, "not shared") +} + +func TestVerifyRequiresCurrentMountPolicyAndPinnedBindings(t *testing.T) { + ctx := context.Background() + rootUID, rootPVCUID, rootPVUID := types.UID("root"), types.UID("root-pvc"), types.UID("root-pv") + leafUID, leafPVCUID, leafPVUID := types.UID("leaf"), types.UID("leaf-pvc"), types.UID("leaf-pv") + target := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "target", Labels: map[string]string{"workspace": "one"}}} + root := &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: "root", Namespace: "root", UID: rootUID}, + Spec: datasetv1alpha1.DatasetSpec{Share: true, ShareAccess: policy(datasetv1alpha1.AccessModeReadWrite, "one"), Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeManual, URI: "manual://"}}, + Status: datasetv1alpha1.DatasetStatus{PVCName: "root-pvc"}, + } + rootPVC := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "root-pvc", Namespace: "root", UID: rootPVCUID}, Spec: corev1.PersistentVolumeClaimSpec{VolumeName: "root-pv"}, Status: corev1.PersistentVolumeClaimStatus{Phase: corev1.ClaimBound}} + rootPV := &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: "root-pv", UID: rootPVUID}, Spec: corev1.PersistentVolumeSpec{ClaimRef: &corev1.ObjectReference{Namespace: "root", Name: "root-pvc", UID: rootPVCUID}, PersistentVolumeSource: corev1.PersistentVolumeSource{NFS: &corev1.NFSVolumeSource{Server: "nfs", Path: "/data"}}}} + leaf := &datasetv1alpha1.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: "leaf", Namespace: "target", UID: leafUID, Generation: 7}, + Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://root/root"}}, + Status: datasetv1alpha1.DatasetStatus{ + PVCName: "leaf-pvc", + ReadOnly: false, + MountSources: []datasetv1alpha1.MountSource{{Namespace: "root", Name: "root", UID: string(rootUID), PVCName: "root-pvc", PVCUID: string(rootPVCUID), PVName: "root-pv", PVUID: string(rootPVUID)}}, + Conditions: []metav1.Condition{{Type: MountPolicyCondition, Status: metav1.ConditionTrue, Reason: "MountPolicyResolved", Message: "", ObservedGeneration: 7}}, + }, + } + owner := *metav1.NewControllerRef(leaf, datasetv1alpha1.GroupVersion.WithKind("Dataset")) + leafPVC := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "leaf-pvc", Namespace: "target", UID: leafPVCUID, OwnerReferences: []metav1.OwnerReference{owner}}, Spec: corev1.PersistentVolumeClaimSpec{VolumeName: "leaf-pv"}, Status: corev1.PersistentVolumeClaimStatus{Phase: corev1.ClaimBound}} + leafPV := &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: "leaf-pv", UID: leafPVUID, OwnerReferences: []metav1.OwnerReference{owner}, Annotations: map[string]string{ + SourceDatasetUIDAnnotation: string(rootUID), SourcePVCUIDAnnotation: string(rootPVCUID), SourcePVUIDAnnotation: string(rootPVUID), + }}, Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "target", Name: "leaf-pvc", UID: leafPVCUID}, + PersistentVolumeSource: corev1.PersistentVolumeSource{NFS: &corev1.NFSVolumeSource{Server: "nfs", Path: "/data"}}, + }} + reader := newReader(t, target, root, rootPVC, rootPV, leafPVC, leafPV) + _, err := Verify(ctx, reader, leaf) + require.NoError(t, err) + + leaf.Status.Conditions[0].ObservedGeneration = 6 + _, err = Verify(ctx, reader, leaf) + require.ErrorContains(t, err, "not ready") +} + +func TestProtectedPVC(t *testing.T) { + ref := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "ref", Namespace: "team"}, Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference}, VolumeClaimTemplate: corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "managed"}}}} + reader := newReader(t, ref) + protected, err := ProtectedPVC(context.Background(), reader, "team", "managed") + require.NoError(t, err) + require.True(t, protected) +}