You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by ya...@apache.org on 2026/10/16 18:19:50 UTC
(airavata-custos) branch master updated: Create storage per user instead of per allocation on each event (#634)
This is an automated email from the ASF dual-hosted git repository.
yasithdev pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
The following commit(s) were added to refs/heads/master by this push:
new 358bee307 Create storage per user instead of per allocation on each event (#634)
358bee307 is described below
commit 358bee3071aca72da0ec87ae5477b394be8d89ce
Author: Lahiru Jayathilake <la...@gmail.com>
AuthorDate: Fri Oct 16 14:19:45 2026 -0400
Create storage per user instead of per allocation on each event (#634)
---
connectors/Storage/VAST-Provisioner/README.md | 10 +-
.../internal/subscribers/subscriber.go | 95 ++++++++++++---
.../internal/subscribers/subscriber_test.go | 131 ++++++++++++++++-----
3 files changed, 182 insertions(+), 54 deletions(-)
diff --git a/connectors/Storage/VAST-Provisioner/README.md b/connectors/Storage/VAST-Provisioner/README.md
index 4a1005dce..304175f76 100644
--- a/connectors/Storage/VAST-Provisioner/README.md
+++ b/connectors/Storage/VAST-Provisioner/README.md
@@ -33,10 +33,12 @@ fields on a view. The policy is looked up once at startup, and Custos does not s
| Trigger | Runs |
|---|---|
-| `compute_cluster_user::update` with `provisioned_at` set | The account's directories, then its project directories |
-| `compute_allocation::create` and `::update` | The project's directories, then each member's; retried until LDAP has the group |
-| `compute_allocation_membership::create` and `::update` | The same, for the membership's allocation |
-| Startup, then every 24 hours | The same, for every account and allocation on each cluster |
+| `compute_cluster_user::provision` | The account's directories, then its directory in each of its projects |
+| `compute_allocation::create` and `::update` | The project's directories; retried until LDAP has the group |
+| `compute_allocation_membership::create` and `::update` | That member's project directory, once their account is provisioned |
+| Startup, then every 24 hours | Everything, for every account and allocation on each cluster |
+
+Each event creates only what it is about, so a failed delivery names the user it is for and can be retried alone.
Configuration is in [`CONFIG.md`](../../../CONFIG.md#storage-vast-provisioner).
diff --git a/connectors/Storage/VAST-Provisioner/internal/subscribers/subscriber.go b/connectors/Storage/VAST-Provisioner/internal/subscribers/subscriber.go
index 804498dbe..630b0ce4d 100644
--- a/connectors/Storage/VAST-Provisioner/internal/subscribers/subscriber.go
+++ b/connectors/Storage/VAST-Provisioner/internal/subscribers/subscriber.go
@@ -69,19 +69,18 @@ func NewStorageSubscriber(bus *events.Bus, core service.CoreService, clusters ma
return &StorageSubscriber{bus: bus, core: core, clusters: clusters}
}
+// Each event creates only its own user's or allocation's mounts, so a failure
+// names the right user and can be retried alone. The reconciler does the rest.
func (s *StorageSubscriber) RegisterSubscribers(subscriber string) {
s.bus.SubscribeComputeClusterUserProvisioned(subscriber, s.ensureUser)
- s.bus.SubscribeComputeAllocationCreated(subscriber, s.ensureAllocation)
- s.bus.SubscribeComputeAllocationUpdated(subscriber, s.ensureAllocation)
- byMembership := func(ctx context.Context, m models.ComputeAllocationMembership) error {
- return s.ensureAllocationByID(ctx, m.ComputeAllocationID)
- }
- s.bus.SubscribeComputeAllocationMembershipCreated(subscriber, byMembership)
- s.bus.SubscribeComputeAllocationMembershipUpdated(subscriber, byMembership)
+ s.bus.SubscribeComputeAllocationCreated(subscriber, s.ensureGroup)
+ s.bus.SubscribeComputeAllocationUpdated(subscriber, s.ensureGroup)
+ s.bus.SubscribeComputeAllocationMembershipCreated(subscriber, s.ensureMember)
+ s.bus.SubscribeComputeAllocationMembershipUpdated(subscriber, s.ensureMember)
}
// ensureUser needs no approval check: COmanage provisions approved accounts only.
-// After the account's mounts it ensures each of the account's allocations.
+// After the account's own mounts, it creates the account's mount in each of its allocations.
func (s *StorageSubscriber) ensureUser(ctx context.Context, cu models.ComputeClusterUser) error {
c, ok := s.clusters[cu.ComputeClusterID]
if !ok || cu.ProvisionedAt == nil {
@@ -94,29 +93,89 @@ func (s *StorageSubscriber) ensureUser(ctx context.Context, cu models.ComputeClu
if err != nil {
return err
}
- var errs []error
for _, m := range memberships {
- errs = append(errs, s.ensureAllocationByID(ctx, m.ComputeAllocationID))
+ if m.MembershipStatus != models.ACTIVE {
+ continue
+ }
+ a, _, err := s.activeAllocation(ctx, m.ComputeAllocationID)
+ if err != nil {
+ return err
+ }
+ if a == nil || a.ComputeClusterID != cu.ComputeClusterID {
+ continue
+ }
+ if err := s.ensureMemberMounts(ctx, c, cu.LocalUsername, *a.PosixGroup); err != nil {
+ return err
+ }
}
- return errors.Join(errs...)
+ return nil
+}
+
+// ensureMember creates the member's mount in the allocation. A member without a
+// provisioned account is skipped, its provision event creates the mount later.
+func (s *StorageSubscriber) ensureMember(ctx context.Context, m models.ComputeAllocationMembership) error {
+ if m.MembershipStatus != models.ACTIVE {
+ return nil
+ }
+ a, c, err := s.activeAllocation(ctx, m.ComputeAllocationID)
+ if err != nil || a == nil {
+ return err
+ }
+ members, err := s.core.ListMembersForAllocation(ctx, a.ID)
+ if err != nil {
+ return err
+ }
+ for _, mw := range members {
+ if mw.UserID == m.UserID && mw.ProvisionedAt != nil {
+ return s.ensureMemberMounts(ctx, c, mw.LocalUsername, *a.PosixGroup)
+ }
+ }
+ return nil
+}
+
+// ensureGroup creates an allocation's group mounts. Members' mounts come from their own events.
+func (s *StorageSubscriber) ensureGroup(ctx context.Context, a models.ComputeAllocation) error {
+ c, ok := s.cluster(a)
+ if !ok {
+ return nil
+ }
+ return s.ensure(ctx, c, "", *a.PosixGroup)
}
-func (s *StorageSubscriber) ensureAllocationByID(ctx context.Context, id string) error {
+// cluster returns the allocation's cluster when the allocation is active, has a group, and the cluster is configured.
+func (s *StorageSubscriber) cluster(a models.ComputeAllocation) (Cluster, bool) {
+ c, ok := s.clusters[a.ComputeClusterID]
+ return c, ok && a.Status == models.ACTIVE && a.PosixGroup != nil
+}
+
+// activeAllocation loads the allocation and its cluster, nil when there is nothing to create for it.
+func (s *StorageSubscriber) activeAllocation(ctx context.Context, id string) (*models.ComputeAllocation, Cluster, error) {
a, err := s.core.GetComputeAllocation(ctx, id)
if errors.Is(err, service.ErrNotFound) {
- return nil
+ return nil, Cluster{}, nil
}
if err != nil {
+ return nil, Cluster{}, err
+ }
+ c, ok := s.cluster(*a)
+ if !ok {
+ return nil, Cluster{}, nil
+ }
+ return a, c, nil
+}
+
+// ensureMemberMounts creates the group's mounts, then the user's mount inside them.
+func (s *StorageSubscriber) ensureMemberMounts(ctx context.Context, c Cluster, user, group string) error {
+ if err := s.ensure(ctx, c, "", group); err != nil {
return err
}
- return s.ensureAllocation(ctx, *a)
+ return s.ensure(ctx, c, user, group)
}
-// ensureAllocation creates an active allocation's mounts, then those of each
-// member its POSIX group holds.
+// ensureAllocation creates an active allocation's mounts, then those of each member its POSIX group holds. Used by the reconciler.
func (s *StorageSubscriber) ensureAllocation(ctx context.Context, a models.ComputeAllocation) error {
- c, ok := s.clusters[a.ComputeClusterID]
- if !ok || a.Status != models.ACTIVE || a.PosixGroup == nil {
+ c, ok := s.cluster(a)
+ if !ok {
return nil
}
if err := s.ensure(ctx, c, "", *a.PosixGroup); err != nil {
diff --git a/connectors/Storage/VAST-Provisioner/internal/subscribers/subscriber_test.go b/connectors/Storage/VAST-Provisioner/internal/subscribers/subscriber_test.go
index 102de5ff7..542fb0fd2 100644
--- a/connectors/Storage/VAST-Provisioner/internal/subscribers/subscriber_test.go
+++ b/connectors/Storage/VAST-Provisioner/internal/subscribers/subscriber_test.go
@@ -39,9 +39,23 @@ var layout = []Mount{
{Path: "/project/{allocation}/{user}", Owner: "{user}", Group: "{user}", Mode: 0o700, View: true},
}
-// oneAllocation serves one active allocation whose only member is the account.
-func oneAllocation() *service.CoreServiceMock {
+var (
+ groupMounts = []string{"folder /project/proj-a", "view /project/proj-a", "quota /project/proj-a"}
+ jdoeOwn = []string{"folder /home/custos-jdoe", "quota /home/custos-jdoe", "folder /scratch/custos-jdoe", "view /scratch/custos-jdoe", "quota /scratch/custos-jdoe"}
+ jdoeMember = []string{"folder /project/proj-a/custos-jdoe", "view /project/proj-a/custos-jdoe"}
+ asmithMember = []string{"folder /project/proj-a/custos-asmith", "view /project/proj-a/custos-asmith"}
+)
+
+// allocationWithMembers serves one active allocation on cluster c1 with two
+// provisioned members, jdoe (u1) and asmith (u2), and one member whose account is not provisioned yet (u3).
+func allocationWithMembers() *service.CoreServiceMock {
group, provisioned := "proj-a", time.Now()
+ member := func(userID, username string, at *time.Time) store.MembershipWithUser {
+ return store.MembershipWithUser{
+ ComputeAllocationMembership: models.ComputeAllocationMembership{ComputeAllocationID: "a1", UserID: userID, MembershipStatus: models.ACTIVE},
+ LocalUsername: username, ProvisionedAt: at,
+ }
+ }
return &service.CoreServiceMock{
ListAllocationsForUserFunc: func(context.Context, string) ([]models.ComputeAllocationMembership, error) {
return []models.ComputeAllocationMembership{{ComputeAllocationID: "a1", MembershipStatus: models.ACTIVE}}, nil
@@ -50,14 +64,43 @@ func oneAllocation() *service.CoreServiceMock {
return &models.ComputeAllocation{ID: "a1", ComputeClusterID: "c1", Status: models.ACTIVE, PosixGroup: &group}, nil
},
ListMembersForAllocationFunc: func(context.Context, string) ([]store.MembershipWithUser, error) {
- return []store.MembershipWithUser{{
- ComputeAllocationMembership: models.ComputeAllocationMembership{MembershipStatus: models.ACTIVE},
- LocalUsername: "custos-jdoe", ProvisionedAt: &provisioned,
- }}, nil
+ return []store.MembershipWithUser{member("u1", "custos-jdoe", &provisioned), member("u2", "custos-asmith", &provisioned), member("u3", "custos-pending", nil)}, nil
},
}
}
+// recordingVMS stands in for VMS and records each create as "<kind> <path>".
+func recordingVMS(t *testing.T) (Cluster, *[]string) {
+ t.Helper()
+ var log []string
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ var body struct{ Path string }
+ _ = json.NewDecoder(r.Body).Decode(&body)
+ switch r.URL.Path {
+ case "/api/token/":
+ _, _ = w.Write([]byte(`{"access":"t"}`))
+ case "/api/folders/create_folder/":
+ log = append(log, "folder "+body.Path)
+ case "/api/quotas/":
+ log = append(log, "quota "+body.Path)
+ case "/api/views/":
+ log = append(log, "view "+body.Path)
+ }
+ }))
+ t.Cleanup(srv.Close)
+ return Cluster{VMS: client.New(srv.URL, "u", "p", 1), Mounts: layout}, &log
+}
+
+func join(parts ...[]string) []string {
+ var all []string
+ for _, p := range parts {
+ all = append(all, p...)
+ }
+ return all
+}
+
+// Make sure a provision event creates the account's own directories and its
+// directory in each of its projects, and nothing for the other members.
func TestEnsureUser(t *testing.T) {
now := time.Now()
tests := []struct {
@@ -66,38 +109,62 @@ func TestEnsureUser(t *testing.T) {
provisionedAt *time.Time
want []string
}{
- {name: "creates in order", cluster: "c1", provisionedAt: &now, want: []string{
- "folder /home/custos-jdoe", "quota /home/custos-jdoe",
- "folder /scratch/custos-jdoe", "view /scratch/custos-jdoe", "quota /scratch/custos-jdoe",
- "folder /project/proj-a", "view /project/proj-a", "quota /project/proj-a",
- "folder /project/proj-a/custos-jdoe", "view /project/proj-a/custos-jdoe",
- }},
+ {name: "own directories, then its project directory", cluster: "c1", provisionedAt: &now, want: join(jdoeOwn, groupMounts, jdoeMember)},
{name: "skips unprovisioned", cluster: "c1"},
{name: "skips unlisted cluster", cluster: "c2", provisionedAt: &now},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- var log []string
- srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
- var body struct{ Path string }
- _ = json.NewDecoder(r.Body).Decode(&body)
- switch r.URL.Path {
- case "/api/token/":
- _, _ = w.Write([]byte(`{"access":"t"}`))
- case "/api/folders/create_folder/":
- log = append(log, "folder "+body.Path)
- case "/api/quotas/":
- log = append(log, "quota "+body.Path)
- case "/api/views/":
- log = append(log, "view "+body.Path)
- }
- }))
- defer srv.Close()
- s := NewStorageSubscriber(nil, oneAllocation(), map[string]Cluster{"c1": {VMS: client.New(srv.URL, "u", "p", 1), Mounts: layout}})
- cu := models.ComputeClusterUser{ComputeClusterID: tt.cluster, LocalUsername: "custos-jdoe", ProvisionedAt: tt.provisionedAt}
- if err := s.ensureUser(context.Background(), cu); err != nil || !slices.Equal(log, tt.want) {
- t.Fatalf("err=%v calls=%v, want %v", err, log, tt.want)
+ c, log := recordingVMS(t)
+ s := NewStorageSubscriber(nil, allocationWithMembers(), map[string]Cluster{"c1": c})
+ cu := models.ComputeClusterUser{ComputeClusterID: tt.cluster, UserID: "u1", LocalUsername: "custos-jdoe", ProvisionedAt: tt.provisionedAt}
+ if err := s.ensureUser(context.Background(), cu); err != nil || !slices.Equal(*log, tt.want) {
+ t.Fatalf("err=%v calls=%v, want %v", err, *log, tt.want)
+ }
+ })
+ }
+}
+
+// Make sure a membership event creates only that member's project directory and nothing until the member's account is provisioned.
+func TestEnsureMember(t *testing.T) {
+ tests := []struct {
+ name, userID string
+ want []string
+ }{
+ {name: "the member's directory only", userID: "u1", want: join(groupMounts, jdoeMember)},
+ {name: "nothing for an unprovisioned member", userID: "u3"},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ c, log := recordingVMS(t)
+ s := NewStorageSubscriber(nil, allocationWithMembers(), map[string]Cluster{"c1": c})
+ m := models.ComputeAllocationMembership{ComputeAllocationID: "a1", UserID: tt.userID, MembershipStatus: models.ACTIVE}
+ if err := s.ensureMember(context.Background(), m); err != nil || !slices.Equal(*log, tt.want) {
+ t.Fatalf("err=%v calls=%v, want %v", err, *log, tt.want)
}
})
}
}
+
+// Make sure an allocation event creates the project's directories and no member's.
+func TestEnsureGroup(t *testing.T) {
+ c, log := recordingVMS(t)
+ group := "proj-a"
+ s := NewStorageSubscriber(nil, allocationWithMembers(), map[string]Cluster{"c1": c})
+ a := models.ComputeAllocation{ID: "a1", ComputeClusterID: "c1", Status: models.ACTIVE, PosixGroup: &group}
+ if err := s.ensureGroup(context.Background(), a); err != nil || !slices.Equal(*log, groupMounts) {
+ t.Fatalf("err=%v calls=%v, want %v", err, *log, groupMounts)
+ }
+}
+
+// Make sure the reconciler's sweep creates the project's directories and every provisioned member's.
+func TestEnsureAllocation(t *testing.T) {
+ c, log := recordingVMS(t)
+ group := "proj-a"
+ s := NewStorageSubscriber(nil, allocationWithMembers(), map[string]Cluster{"c1": c})
+ a := models.ComputeAllocation{ID: "a1", ComputeClusterID: "c1", Status: models.ACTIVE, PosixGroup: &group}
+ want := join(groupMounts, jdoeMember, asmithMember)
+ if err := s.ensureAllocation(context.Background(), a); err != nil || !slices.Equal(*log, want) {
+ t.Fatalf("err=%v calls=%v, want %v", err, *log, want)
+ }
+}