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/01 17:11:39 UTC
(airavata-custos) branch vast-storage updated (9ff04257e -> 88377d9c0)
This is an automated email from the ASF dual-hosted git repository.
yasithdev pushed a change to branch vast-storage
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
discard 9ff04257e base implementation
add 34dfb7ee5 remove unneede if conditions
add 38c1be742 remove unused pieces
add f4c010909 reuse existing functions wherever possible
new 88377d9c0 create vast storage records when user or allocation is provisioned
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (9ff04257e)
\
N -- N -- N refs/heads/vast-storage (88377d9c0)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
config/custos.yaml | 8 +----
.../internal/client/group_members.go | 16 +--------
.../Identity-Provisioner/internal/client/groups.go | 7 +++-
.../internal/operations/ensure_cluster_admin.go | 18 ++--------
.../internal/operations/lookup.go | 25 ++++++++++++++
.../internal/operations/orchestrator.go | 26 ++++-----------
.../internal/subscribers/cluster_user.go | 8 ++---
.../Identity-Provisioner/pkg/comanage/loader.go | 7 +---
.../internal/subscribers/members.go | 16 ---------
.../internal/subscribers/subscriber.go | 1 -
connectors/VAST/Storage-Provisioner/README.md | 6 ++--
.../Storage-Provisioner/internal/client/client.go | 37 +++++++++-----------
.../internal/client/client_test.go | 13 ++++----
.../internal/operations/ensure.go | 39 ++++++----------------
.../internal/operations/ensure_test.go | 20 ++++-------
.../internal/subscribers/subscriber.go | 27 +++++----------
.../VAST/Storage-Provisioner/pkg/vast/loader.go | 3 +-
.../Storage-Provisioner/pkg/vast/loader_test.go | 13 ++------
internal/store/compute_allocation_store.go | 9 ++---
internal/store/store.go | 4 +--
pkg/posix/username.go | 13 ++------
pkg/service/compute_allocation.go | 11 +++---
22 files changed, 106 insertions(+), 221 deletions(-)
(airavata-custos) 01/01: create vast storage records when user or allocation is provisioned
Posted by ya...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yasithdev pushed a commit to branch vast-storage
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit 88377d9c0964e12c05d5f714182c7e8f878d8976
Author: Yasith Jayawardana <ya...@icloud.com>
AuthorDate: Thu Oct 1 13:11:26 2026 -0400
create vast storage records when user or allocation is provisioned
---
CONFIG.md | 27 +++
config/custos.yaml | 38 +++++
connectors/VAST/Storage-Provisioner/README.md | 79 +++++++++
.../Storage-Provisioner/internal/client/client.go | 182 +++++++++++++++++++++
.../internal/client/client_test.go | 100 +++++++++++
.../internal/operations/ensure.go | 164 +++++++++++++++++++
.../internal/operations/ensure_test.go | 130 +++++++++++++++
.../internal/subscribers/subscriber.go | 123 ++++++++++++++
.../VAST/Storage-Provisioner/pkg/vast/loader.go | 121 ++++++++++++++
.../Storage-Provisioner/pkg/vast/loader_test.go | 54 ++++++
internal/connectors/loader.go | 2 +
11 files changed, 1020 insertions(+)
diff --git a/CONFIG.md b/CONFIG.md
index 91915a61e..1fe445b4f 100644
--- a/CONFIG.md
+++ b/CONFIG.md
@@ -132,6 +132,32 @@ connectors:
http_timeout: "30s"
```
+#### VAST Storage Provisioner
+
+```yaml
+ vast-storage:
+ type: "vast-storage-provisioner"
+ enabled: true
+ custos_cluster_id: "cluster-001"
+ vms:
+ url: "https://vms.example.org"
+ username: "custos"
+ password: "${VAST_API_PASSWORD}"
+ tenant_id: 1
+ # For a VMS on its self-signed default certificate: trust it and check its name.
+ # ca_file: "/etc/custos/vast-ca.pem"
+ # server_name: "vms.vastdata.com"
+ # One entry per directory; {user} and {allocation} in path, owner and group decide whether it
+ # waits for the account, the allocation's group, or both. config/custos.yaml holds the full list.
+ storage:
+ - path: "/home/{user}"
+ owner: "{user}"
+ group: "{user}"
+ mode: "0700"
+ hard_limit: 107374182400 # 100 GiB
+ hard_limit_inodes: 1000000
+```
+
#### AMIE Processor
```yaml
@@ -223,6 +249,7 @@ When a connector is disabled:
- `slurm-association-mapper` - SLURM Association Mapper
- `slurm-usage-monitor` - SLURM Usage Monitor
- `comanage-identity-provisioner` - COmanage Identity Provisioner
+- `vast-storage-provisioner` - VAST Storage Provisioner
- `amie-processor` - AMIE Processor
New connector types can be added by:
diff --git a/config/custos.yaml b/config/custos.yaml
index 53684f48e..39f0a756c 100644
--- a/config/custos.yaml
+++ b/config/custos.yaml
@@ -77,6 +77,44 @@ connectors:
homedir_prefix: "/home/"
http_timeout: "30s"
+ vast-storage:
+ type: "vast-storage-provisioner"
+ enabled: false
+ custos_cluster_id: "${CUSTOS_CLUSTER_ID}"
+ vms:
+ url: "${VAST_VMS_URL}"
+ username: "${VAST_API_USER}"
+ password: "${VAST_API_PASSWORD}"
+ tenant_id: ${VAST_TENANT_ID}
+ # allowed string variables: {user}, {allocation}
+ storage:
+ - path: "/home/{user}"
+ owner: "{user}"
+ group: "{user}"
+ mode: "0700"
+ hard_limit: 107374182400 # 100 GiB
+ hard_limit_inodes: 1000000
+ - path: "/scratch/{user}"
+ owner: "{user}"
+ group: "{user}"
+ mode: "0700"
+ hard_limit: 10995116277760 # 10 TiB
+ hard_limit_inodes: 10000000
+ - path: "/project/{allocation}"
+ owner: "0"
+ group: "{allocation}"
+ mode: "2750"
+ hard_limit: 10995116277760 # 10 TiB
+ hard_limit_inodes: 300000000
+ - path: "/project/{allocation}/shared"
+ owner: "0"
+ group: "{allocation}"
+ mode: "2770"
+ - path: "/project/{allocation}/{user}"
+ owner: "{user}"
+ group: "{user}"
+ mode: "0700"
+
amie-processor:
type: "amie-processor"
enabled: true
diff --git a/connectors/VAST/Storage-Provisioner/README.md b/connectors/VAST/Storage-Provisioner/README.md
new file mode 100644
index 000000000..5c42fe563
--- /dev/null
+++ b/connectors/VAST/Storage-Provisioner/README.md
@@ -0,0 +1,79 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+
+# VAST Storage-Provisioner
+
+Creates directories and quotas on a VAST cluster through the VMS REST API, once the COmanage
+Identity-Provisioner on the same Custos cluster has provisioned the account or the allocation's group. VAST resolves
+owners by name from the LDAP COmanage feeds. Nothing is changed once it exists, and nothing is deleted.
+
+The directories come from the `storage` list in the config. Each entry's `path`, `owner` and `group` are templates
+over `{user}`, the account's username, and `{allocation}`, the allocation's group, and the placeholders it uses decide
+what it waits for: the account, the group, or both. Entries are created in config order, with a quota named after the
+path when `hard_limit` is set. The shipped config holds the Nexus policy:
+
+| Uses | Created when | Path | Owner and mode | Quota |
+|---|---|---|---|---|
+| `{user}` | The account is approved and provisioned | `/home/{user}` | `{user}:{user} 0700` | 100 GiB, 1M inodes |
+| `{user}` | The account is approved and provisioned | `/scratch/{user}` | `{user}:{user} 0700` | 10 TiB, 10M inodes |
+| `{allocation}` | The allocation is `ACTIVE` and its group provisioned | `/project/{allocation}` | `0:{allocation} 2750` | 10 TiB, 300M inodes |
+| `{allocation}` | The allocation is `ACTIVE` and its group provisioned | `/project/{allocation}/shared` | `0:{allocation} 2770` | Counts against the project |
+| Both | Both, and the user is in the group | `/project/{allocation}/{user}` | `{user}:{user} 0700` | Counts against the project |
+
+| Nexus storage policy | Enforced by |
+|---|---|
+| Home, scratch and project quotas | The connector, as above |
+| More project quota on request | An admin raises it in VMS; the connector never changes a quota that exists |
+| Home 1 TiB where the project space is shared only | Config: drop `/project/{allocation}/{user}` and set home to 1 TiB |
+| Home in the Home tenant, the rest in the Data tenant | Not yet: Nexus has one tenant, `vms.tenant_id` |
+| Snapshots, no backup, scratch data deleted after 30 days, trial expiry | VAST view policies, set by admins |
+| Data, trial and software views | Admins; they belong to no account or allocation |
+| Purge at the end of an allocation and its grace period | Admins; VMS refuses folder deletes while the trash folder is disabled |
+
+| Trigger | Runs |
+|---|---|
+| `compute_cluster_user::update` | `EnsureUser` |
+| `compute_allocation::update` | `EnsureAllocation` |
+| `compute_allocation_membership::create` and `::update` | `EnsureUser` for the user's account on the cluster |
+| Startup, then every 24 hours | `EnsureUser` and `EnsureAllocation` for everything on the cluster |
+
+| VMS response | Handling |
+|---|---|
+| Already exists (`503 ALREADY_EXISTS`, or a taken quota name) | Success, once the folder or quota is found |
+| 401, or 403 `authentication_failed` | Log in again, once |
+| `503 Couldn't find user` or `group` | Retried by the bus while the directory catches up |
+| Other 503 with a JSON `code` | Retried by the bus |
+| A timeout, or a 5xx without a JSON `code` | Retried by the bus; calls fail at once for 60 seconds |
+| Other 4xx | `events.ErrPermanent` |
+
+Configuration is in [`CONFIG.md`](../../../CONFIG.md#vast-storage-provisioner).
+
+## Prerequisites on VMS
+
+Found on the Nexus VMS on 2026-10-01; PACE admins make these changes before the connector is enabled.
+
+| Prerequisite | Why |
+|---|---|
+| The COmanage LDAP is a VMS user-directory provider | VMS resolves owners only from its providers, today GT's AD (`gted.gatech.edu`); `custos-*` users and `proj-*` groups are otherwise unknown and every create retries |
+| The setgid bit is kept on `<project>/<group>` and `shared` | VMS created a folder requested as `2770` as `0770`, so files in `shared` would not inherit `<group>` |
+| The service account may create folders and quotas | The account the connector logs in with |
+| A certificate for the VMS hostname, or `ca_file` and `server_name: vms.vastdata.com` | VMS serves VAST's self-signed default certificate |
+
+Folders cannot be deleted through the API while the
+trash folder is disabled.
diff --git a/connectors/VAST/Storage-Provisioner/internal/client/client.go b/connectors/VAST/Storage-Provisioner/internal/client/client.go
new file mode 100644
index 000000000..8f552515b
--- /dev/null
+++ b/connectors/VAST/Storage-Provisioner/internal/client/client.go
@@ -0,0 +1,182 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package client creates folders and quotas through the VAST VMS REST API.
+// Auth is a JWT from `/api/token/`, renewed once on a 401. VMS answers most
+// refusals with a 503 carrying a JSON `code`, so a refused create counts as
+// success when the folder or quota exists, an unknown owner stays retryable
+// while the directory catches up, and any other 4xx is permanent. Only an
+// outage (a timeout, or a 5xx without that JSON) makes calls fail at once for
+// 60 seconds, so it does not hold the bus.
+package client
+
+import (
+ "bytes"
+ "context"
+ "crypto/tls"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "regexp"
+ "strings"
+ "time"
+
+ "github.com/apache/airavata-custos/pkg/events"
+)
+
+type Client struct {
+ url, username, password string
+ tenantID int
+ http *http.Client
+ token string
+ pausedUntil time.Time
+}
+
+func New(vmsURL, username, password string, tenantID int, tlsConfig *tls.Config) *Client {
+ return &Client{
+ url: strings.TrimSuffix(vmsURL, "/") + "/api/",
+ username: username,
+ password: password,
+ tenantID: tenantID,
+ http: &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{TLSClientConfig: tlsConfig}},
+ }
+}
+
+// CreateFolder creates path owned by user:group unless it exists.
+func (c *Client) CreateFolder(ctx context.Context, path, user, group string, mode int) error {
+ return c.create(ctx, "folders/create_folder/", map[string]any{
+ "path": path, "user": user, "group": group, "owner_is_group": false, "create_dir_mode": fmt.Sprintf("%o", mode), "tenant_id": c.tenantID,
+ }, func() bool {
+ return c.do(ctx, http.MethodPost, "folders/stat_path/", map[string]any{"path": path, "tenant_id": c.tenantID}, nil) == nil
+ })
+}
+
+// CreateQuota creates the quota name on path unless a quota by that name exists.
+func (c *Client) CreateQuota(ctx context.Context, name, path string, hardLimit, hardLimitInodes int64) error {
+ return c.create(ctx, "quotas/", map[string]any{
+ "name": name, "path": path, "create_dir": false, "hard_limit": hardLimit, "hard_limit_inodes": hardLimitInodes, "tenant_id": c.tenantID,
+ }, func() bool {
+ var found []json.RawMessage
+ return c.do(ctx, http.MethodGet, "quotas/?name="+url.QueryEscape(name), nil, &found) == nil && len(found) > 0
+ })
+}
+
+var unknownOwner = regexp.MustCompile(`Couldn't find (user|group)`)
+
+// authFailed matches the 403 codes VMS gives a bad password and a bad or expired token.
+var authFailed = regexp.MustCompile(`"code":"(authentication_failed|token_not_valid)"`)
+
+// create posts body. exists calls do, not call, since VMS answers a missing
+// path with a 503 that is not an outage.
+func (c *Client) create(ctx context.Context, path string, body map[string]any, exists func() bool) error {
+ err := c.call(ctx, http.MethodPost, path, body, nil)
+ if err == nil || !isVMSAnswer(err) {
+ return err
+ }
+ if exists() {
+ return nil
+ }
+ var he *httpError
+ if unknownOwner.MatchString(err.Error()) || (errors.As(err, &he) && he.status >= 500) {
+ return err
+ }
+ return fmt.Errorf("%w: %v", events.ErrPermanent, err)
+}
+
+var (
+ errUnauthorized = errors.New("vast: 401 unauthorized")
+ errPaused = errors.New("vast: paused after a VMS failure")
+)
+
+type httpError struct {
+ status int
+ msg string
+}
+
+func (e *httpError) Error() string { return fmt.Sprintf("vast: %d: %s", e.status, e.msg) }
+
+// isVMSAnswer reports whether VMS itself answered: a 4xx, or a 5xx with its
+// JSON error `code`. Anything else is an outage.
+func isVMSAnswer(err error) bool {
+ var he *httpError
+ return errors.As(err, &he) && (he.status < 500 || strings.Contains(he.msg, `"code"`))
+}
+
+func (c *Client) call(ctx context.Context, method, path string, body, out any) error {
+ if time.Now().Before(c.pausedUntil) {
+ return errPaused
+ }
+ err := errUnauthorized
+ if c.token != "" {
+ err = c.do(ctx, method, path, body, out)
+ }
+ if errors.Is(err, errUnauthorized) {
+ if err = c.login(ctx); err == nil {
+ err = c.do(ctx, method, path, body, out)
+ }
+ }
+ if err != nil && !isVMSAnswer(err) && !errors.Is(err, errUnauthorized) {
+ c.pausedUntil = time.Now().Add(60 * time.Second)
+ }
+ return err
+}
+
+func (c *Client) login(ctx context.Context) error {
+ c.token = ""
+ var t struct{ Access string }
+ err := c.do(ctx, http.MethodPost, "token/", map[string]string{"username": c.username, "password": c.password}, &t)
+ c.token = t.Access
+ return err
+}
+
+func (c *Client) do(ctx context.Context, method, path string, body, out any) error {
+ var rdr io.Reader
+ if body != nil {
+ b, err := json.Marshal(body)
+ if err != nil {
+ return err
+ }
+ rdr = bytes.NewReader(b)
+ }
+ req, err := http.NewRequestWithContext(ctx, method, c.url+path, rdr)
+ if err != nil {
+ return err
+ }
+ req.Header.Set("Content-Type", "application/json")
+ if c.token != "" {
+ req.Header.Set("Authorization", "Bearer "+c.token)
+ }
+ resp, err := c.http.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode >= 300 {
+ msg, _ := io.ReadAll(io.LimitReader(resp.Body, 300))
+ if resp.StatusCode == http.StatusUnauthorized || authFailed.Match(msg) {
+ return errUnauthorized
+ }
+ return &httpError{resp.StatusCode, string(msg)}
+ }
+ if out == nil {
+ return nil
+ }
+ return json.NewDecoder(resp.Body).Decode(out)
+}
diff --git a/connectors/VAST/Storage-Provisioner/internal/client/client_test.go b/connectors/VAST/Storage-Provisioner/internal/client/client_test.go
new file mode 100644
index 000000000..0afc46a4c
--- /dev/null
+++ b/connectors/VAST/Storage-Provisioner/internal/client/client_test.go
@@ -0,0 +1,100 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package client
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+ "time"
+
+ "github.com/apache/airavata-custos/pkg/events"
+)
+
+// vms issues token "fresh" and answers every other call with create's status
+// and body, or stat's status for stat_path.
+func vms(t *testing.T, create int, body string, stat int, calls *int) *Client {
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ switch {
+ case r.URL.Path == "/api/token/":
+ _, _ = w.Write([]byte(`{"access":"fresh","refresh":"r"}`))
+ case r.Header.Get("Authorization") != "Bearer fresh":
+ w.WriteHeader(http.StatusForbidden)
+ _, _ = w.Write([]byte(`{"detail":"Given token not valid for any token type","code":"token_not_valid"}`))
+ case r.URL.Path == "/api/folders/stat_path/":
+ w.WriteHeader(stat)
+ default:
+ *calls++
+ if create == 0 {
+ time.Sleep(50 * time.Millisecond)
+ return
+ }
+ w.WriteHeader(create)
+ _, _ = w.Write([]byte(body))
+ }
+ }))
+ t.Cleanup(srv.Close)
+ c := New(srv.URL, "u", "p", 1, nil)
+ c.token = "expired"
+ c.http.Timeout = 20 * time.Millisecond
+ return c
+}
+
+func TestCreateFolder(t *testing.T) {
+ // Bodies are what the Nexus VMS returned on 2026-10-01.
+ tests := []struct {
+ name string
+ create int
+ body string
+ stat int
+ wantPermanent bool
+ wantErr bool
+ wantPause bool
+ }{
+ {name: "created", create: 200},
+ {name: "already exists", create: 503, body: `{"detail":"create_directory returned an error : CreateDirResultCode.ALREADY_EXISTS","code":"service_unavailable"}`, stat: 200},
+ {name: "unknown owner", create: 503, body: `{"detail":"Couldn't find user with username=custos-jdoe","code":"service_unavailable"}`, stat: 503, wantErr: true},
+ {name: "other 4xx", create: 400, body: `{"detail":"bad","code":"bad_request"}`, stat: 503, wantErr: true, wantPermanent: true},
+ {name: "outage", create: 502, body: "Bad Gateway", wantErr: true, wantPause: true},
+ {name: "timeout", create: 0, wantErr: true, wantPause: true},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ var calls int
+ c := vms(t, tt.create, tt.body, tt.stat, &calls)
+ err := c.CreateFolder(context.Background(), "/home/custos-jdoe", "custos-jdoe", "custos-jdoe", 0o700)
+ if (err != nil) != tt.wantErr || errors.Is(err, events.ErrPermanent) != tt.wantPermanent {
+ t.Fatalf("err=%v", err)
+ }
+ if c.pausedUntil.IsZero() == tt.wantPause {
+ t.Fatalf("paused=%v, want %v", !c.pausedUntil.IsZero(), tt.wantPause)
+ }
+ })
+ }
+}
+
+func TestPauseAfterFailure(t *testing.T) {
+ var calls int
+ c := vms(t, 502, "", 404, &calls)
+ _ = c.CreateQuota(context.Background(), "proj-a", "/project/proj-a", 1, 1)
+ if err := c.CreateQuota(context.Background(), "proj-a", "/project/proj-a", 1, 1); !errors.Is(err, errPaused) || calls != 1 {
+ t.Fatalf("second call must not reach VMS: err=%v calls=%d", err, calls)
+ }
+}
diff --git a/connectors/VAST/Storage-Provisioner/internal/operations/ensure.go b/connectors/VAST/Storage-Provisioner/internal/operations/ensure.go
new file mode 100644
index 000000000..da4b3c2b5
--- /dev/null
+++ b/connectors/VAST/Storage-Provisioner/internal/operations/ensure.go
@@ -0,0 +1,164 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package operations creates the configured directories and quotas on VAST
+// once COmanage has provisioned what each depends on: an account, an
+// allocation's group, or both, since VAST resolves owners by name from the LDAP
+// that COmanage feeds. Mounts are created in config order, so parents come
+// first. Nothing is changed once it exists, and nothing is deleted. Every step
+// returns on its first error, and a retry resumes where it stopped.
+package operations
+
+import (
+ "context"
+ "errors"
+ "slices"
+ "strings"
+
+ "github.com/apache/airavata-custos/connectors/VAST/Storage-Provisioner/internal/client"
+ "github.com/apache/airavata-custos/pkg/models"
+ "github.com/apache/airavata-custos/pkg/service"
+)
+
+// CoreService is the subset of pkg/service.Service the operations depend on.
+type CoreService interface {
+ GetComputeAllocation(ctx context.Context, id string) (*models.ComputeAllocation, error)
+ ListAllocationsForUser(ctx context.Context, userID string) ([]models.ComputeAllocationMembership, error)
+ ListGroupMembersForAllocation(ctx context.Context, allocationID string) ([]models.ComputeClusterUser, error)
+}
+
+// Mount is a directory created for each account, each allocation's group, or
+// each member of a group. Path, Owner and Group are templates over {user}, the
+// account's username, and {allocation}, the allocation's group. A quota named
+// after the path is created when HardLimit is set.
+type Mount struct {
+ Path, Owner, Group string
+ Mode int
+ HardLimit, HardLimitInodes int64
+}
+
+// DependsOn reads what the mount waits for from the placeholders it uses, or ""
+// when it uses neither.
+func (m Mount) DependsOn() string {
+ t := m.Path + m.Owner + m.Group
+ switch user, alloc := strings.Contains(t, "{user}"), strings.Contains(t, "{allocation}"); {
+ case user && alloc:
+ return "both"
+ case user:
+ return "user"
+ case alloc:
+ return "allocation"
+ }
+ return ""
+}
+
+type Operations struct {
+ vms *client.Client
+ core CoreService
+ custosClusterID string
+ mounts []Mount
+}
+
+func New(vms *client.Client, core CoreService, custosClusterID string, mounts []Mount) *Operations {
+ return &Operations{vms: vms, core: core, custosClusterID: custosClusterID, mounts: mounts}
+}
+
+// EnsureUser creates a provisioned account's mounts, then its mounts in each
+// provisioned group that holds it. COmanage provisions approved accounts only.
+func (o *Operations) EnsureUser(ctx context.Context, cu *models.ComputeClusterUser) error {
+ if cu.ComputeClusterID != o.custosClusterID || cu.ProvisionedAt == nil {
+ return nil
+ }
+ if err := o.ensure(ctx, "user", cu.LocalUsername, ""); err != nil {
+ return err
+ }
+ memberships, err := o.core.ListAllocationsForUser(ctx, cu.UserID)
+ if err != nil {
+ return err
+ }
+ for _, m := range memberships {
+ a, err := o.core.GetComputeAllocation(ctx, m.ComputeAllocationID)
+ if errors.Is(err, service.ErrNotFound) {
+ continue
+ }
+ if err != nil {
+ return err
+ }
+ if !o.provisioned(a) {
+ continue
+ }
+ members, err := o.core.ListGroupMembersForAllocation(ctx, a.ID)
+ if err != nil {
+ return err
+ }
+ if !slices.ContainsFunc(members, func(m models.ComputeClusterUser) bool { return m.ID == cu.ID }) {
+ continue
+ }
+ if err := o.ensureAllocation(ctx, a, *cu); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// EnsureAllocation creates a provisioned group's mounts, then each member's.
+func (o *Operations) EnsureAllocation(ctx context.Context, a *models.ComputeAllocation) error {
+ if !o.provisioned(a) {
+ return nil
+ }
+ members, err := o.core.ListGroupMembersForAllocation(ctx, a.ID)
+ if err != nil {
+ return err
+ }
+ return o.ensureAllocation(ctx, a, members...)
+}
+
+func (o *Operations) provisioned(a *models.ComputeAllocation) bool {
+ return a.ComputeClusterID == o.custosClusterID && a.Status == models.ACTIVE && a.PosixGroup != nil && a.PosixGroupProvisionedAt != nil
+}
+
+func (o *Operations) ensureAllocation(ctx context.Context, a *models.ComputeAllocation, members ...models.ComputeClusterUser) error {
+ if err := o.ensure(ctx, "allocation", "", *a.PosixGroup); err != nil {
+ return err
+ }
+ for _, cu := range members {
+ if err := o.ensure(ctx, "both", cu.LocalUsername, *a.PosixGroup); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func (o *Operations) ensure(ctx context.Context, kind, user, group string) error {
+ r := strings.NewReplacer("{user}", user, "{allocation}", group)
+ for _, m := range o.mounts {
+ if m.DependsOn() != kind {
+ continue
+ }
+ path := r.Replace(m.Path)
+ if err := o.vms.CreateFolder(ctx, path, r.Replace(m.Owner), r.Replace(m.Group), m.Mode); err != nil {
+ return err
+ }
+ if m.HardLimit == 0 {
+ continue
+ }
+ if err := o.vms.CreateQuota(ctx, path, path, m.HardLimit, m.HardLimitInodes); err != nil {
+ return err
+ }
+ }
+ return nil
+}
diff --git a/connectors/VAST/Storage-Provisioner/internal/operations/ensure_test.go b/connectors/VAST/Storage-Provisioner/internal/operations/ensure_test.go
new file mode 100644
index 000000000..08125f7a4
--- /dev/null
+++ b/connectors/VAST/Storage-Provisioner/internal/operations/ensure_test.go
@@ -0,0 +1,130 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package operations
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "slices"
+ "testing"
+ "time"
+
+ "github.com/apache/airavata-custos/connectors/VAST/Storage-Provisioner/internal/client"
+ "github.com/apache/airavata-custos/pkg/models"
+)
+
+type fakeCore struct {
+ alloc models.ComputeAllocation
+ members []models.ComputeClusterUser
+}
+
+func (f *fakeCore) GetComputeAllocation(context.Context, string) (*models.ComputeAllocation, error) {
+ return &f.alloc, nil
+}
+func (f *fakeCore) ListAllocationsForUser(context.Context, string) ([]models.ComputeAllocationMembership, error) {
+ return []models.ComputeAllocationMembership{{ComputeAllocationID: f.alloc.ID}}, nil
+}
+func (f *fakeCore) ListGroupMembersForAllocation(context.Context, string) ([]models.ComputeClusterUser, error) {
+ return f.members, nil
+}
+
+// recordingVMS logs each create as "folder <path>" or "quota <path>", and
+// fails the create of failPath.
+func recordingVMS(t *testing.T, failPath string) (*client.Client, *[]string) {
+ 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"}`))
+ return
+ case "/api/folders/create_folder/":
+ log = append(log, "folder "+body.Path)
+ case "/api/quotas/":
+ log = append(log, "quota "+body.Path)
+ }
+ if body.Path == failPath {
+ w.WriteHeader(http.StatusServiceUnavailable)
+ }
+ }))
+ t.Cleanup(srv.Close)
+ return client.New(srv.URL, "u", "p", 1, nil), &log
+}
+
+// layout mirrors the Nexus config, but leaves scratch on the view default.
+var layout = []Mount{
+ {Path: "/home/{user}", Owner: "{user}", Group: "{user}", Mode: 0o700, HardLimit: 1, HardLimitInodes: 1},
+ {Path: "/scratch/{user}", Owner: "{user}", Group: "{user}", Mode: 0o700},
+ {Path: "/project/{allocation}", Owner: "0", Group: "{allocation}", Mode: 0o2750, HardLimit: 1, HardLimitInodes: 1},
+ {Path: "/project/{allocation}/shared", Owner: "0", Group: "{allocation}", Mode: 0o2770},
+ {Path: "/project/{allocation}/{user}", Owner: "{user}", Group: "{user}", Mode: 0o700},
+}
+
+func fixture() (*models.ComputeClusterUser, *fakeCore) {
+ now := time.Now()
+ group := "proj-a"
+ cu := &models.ComputeClusterUser{ID: "cu1", UserID: "u1", ComputeClusterID: "c1", LocalUsername: "custos-jdoe",
+ ApprovalStatus: models.ClusterAccountApproved, ProvisionedAt: &now}
+ core := &fakeCore{
+ alloc: models.ComputeAllocation{ID: "a1", ComputeClusterID: "c1", Status: models.ACTIVE, PosixGroup: &group, PosixGroupProvisionedAt: &now},
+ members: []models.ComputeClusterUser{*cu},
+ }
+ return cu, core
+}
+
+func TestEnsureUser(t *testing.T) {
+ cu, core := fixture()
+ vms, log := recordingVMS(t, "")
+ if err := New(vms, core, "c1", layout).EnsureUser(context.Background(), cu); err != nil {
+ t.Fatal(err)
+ }
+ want := []string{"folder /home/custos-jdoe", "quota /home/custos-jdoe", "folder /scratch/custos-jdoe", "folder /project/proj-a",
+ "quota /project/proj-a", "folder /project/proj-a/shared", "folder /project/proj-a/custos-jdoe"}
+ if !slices.Equal(*log, want) {
+ t.Fatalf("got %v, want %v", *log, want)
+ }
+}
+
+func TestEnsureUser_SkipsUnprovisioned(t *testing.T) {
+ cu, core := fixture()
+ cu.ProvisionedAt = nil
+ vms, log := recordingVMS(t, "")
+ if err := New(vms, core, "c1", layout).EnsureUser(context.Background(), cu); err != nil || len(*log) > 0 {
+ t.Fatalf("err=%v calls=%v", err, *log)
+ }
+}
+
+func TestEnsureAllocation_SkipsUnprovisionedGroup(t *testing.T) {
+ _, core := fixture()
+ core.alloc.PosixGroupProvisionedAt = nil
+ vms, log := recordingVMS(t, "")
+ if err := New(vms, core, "c1", layout).EnsureAllocation(context.Background(), &core.alloc); err != nil || len(*log) > 0 {
+ t.Fatalf("err=%v calls=%v", err, *log)
+ }
+}
+
+func TestEnsureAllocation_StopsAtFirstError(t *testing.T) {
+ _, core := fixture()
+ vms, log := recordingVMS(t, "/project/proj-a")
+ if err := New(vms, core, "c1", layout).EnsureAllocation(context.Background(), &core.alloc); err == nil || len(*log) != 1 {
+ t.Fatalf("err=%v calls=%v", err, *log)
+ }
+}
diff --git a/connectors/VAST/Storage-Provisioner/internal/subscribers/subscriber.go b/connectors/VAST/Storage-Provisioner/internal/subscribers/subscriber.go
new file mode 100644
index 000000000..637b04c68
--- /dev/null
+++ b/connectors/VAST/Storage-Provisioner/internal/subscribers/subscriber.go
@@ -0,0 +1,123 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package subscribers
+
+import (
+ "context"
+ "errors"
+ "log/slog"
+ "sync"
+ "time"
+
+ "github.com/apache/airavata-custos/connectors/VAST/Storage-Provisioner/internal/operations"
+ "github.com/apache/airavata-custos/pkg/events"
+ "github.com/apache/airavata-custos/pkg/models"
+ "github.com/apache/airavata-custos/pkg/service"
+)
+
+// StorageSubscriber keeps the directories on VAST for `CustosClusterID` in
+// step with Custos, ignoring every other cluster.
+type StorageSubscriber struct {
+ ops *operations.Operations
+ bus *events.Bus
+ core *service.Service
+ custosClusterID string
+ mu sync.Mutex // serializes the reconciler with event deliveries
+}
+
+func NewStorageSubscriber(ops *operations.Operations, bus *events.Bus, core *service.Service, custosClusterID string) *StorageSubscriber {
+ return &StorageSubscriber{ops: ops, bus: bus, core: core, custosClusterID: custosClusterID}
+}
+
+// RegisterSubscribers subscribes the handlers under the given subscriber name.
+// The "provisioned" updates COmanage publishes are what release the work.
+func (s *StorageSubscriber) RegisterSubscribers(subscriber string) {
+ s.bus.SubscribeComputeClusterUserUpdated(subscriber, func(ctx context.Context, cu models.ComputeClusterUser) error {
+ return s.ensureUser(ctx, cu.ID)
+ })
+ s.bus.SubscribeComputeAllocationUpdated(subscriber, func(ctx context.Context, a models.ComputeAllocation) error {
+ return s.ensureAllocation(ctx, a.ID)
+ })
+ byMembership := func(ctx context.Context, m models.ComputeAllocationMembership) error {
+ cu, err := s.core.GetComputeClusterUserByPair(ctx, s.custosClusterID, m.UserID)
+ if errors.Is(err, service.ErrNotFound) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ return s.ensureUser(ctx, cu.ID)
+ }
+ s.bus.SubscribeComputeAllocationMembershipCreated(subscriber, byMembership)
+ s.bus.SubscribeComputeAllocationMembershipUpdated(subscriber, byMembership)
+}
+
+// ensureUser and ensureAllocation reload the entity under the lock, since a
+// payload may be stale.
+func (s *StorageSubscriber) ensureUser(ctx context.Context, id string) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ cu, err := s.core.GetComputeClusterUser(ctx, id)
+ if errors.Is(err, service.ErrNotFound) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ return s.ops.EnsureUser(ctx, cu)
+}
+
+func (s *StorageSubscriber) ensureAllocation(ctx context.Context, id string) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ a, err := s.core.GetComputeAllocation(ctx, id)
+ if errors.Is(err, service.ErrNotFound) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ return s.ops.EnsureAllocation(ctx, a)
+}
+
+func (s *StorageSubscriber) StartReconciler(ctx context.Context) {
+ for {
+ s.reconcile(ctx)
+ select {
+ case <-ctx.Done():
+ return
+ case <-time.After(24 * time.Hour):
+ }
+ }
+}
+
+func (s *StorageSubscriber) reconcile(ctx context.Context) {
+ users, err := s.core.ListComputeClusterUsersByCluster(ctx, s.custosClusterID)
+ errs := []error{err}
+ for _, cu := range users {
+ errs = append(errs, s.ensureUser(ctx, cu.ID))
+ }
+ allocs, err := s.core.ListComputeAllocationsByCluster(ctx, s.custosClusterID)
+ errs = append(errs, err)
+ for _, a := range allocs {
+ errs = append(errs, s.ensureAllocation(ctx, a.ID))
+ }
+ if err := errors.Join(errs...); err != nil {
+ slog.Error("VAST reconciler", "err", err)
+ }
+}
diff --git a/connectors/VAST/Storage-Provisioner/pkg/vast/loader.go b/connectors/VAST/Storage-Provisioner/pkg/vast/loader.go
new file mode 100644
index 000000000..3ab57d5b6
--- /dev/null
+++ b/connectors/VAST/Storage-Provisioner/pkg/vast/loader.go
@@ -0,0 +1,121 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package vast is the VAST Storage-Provisioner entry point. Wired from
+// internal/connectors/loader.go.
+package vast
+
+import (
+ "context"
+ "crypto/tls"
+ "crypto/x509"
+ "fmt"
+ "log/slog"
+ "os"
+ "strconv"
+ "sync"
+
+ "github.com/jmoiron/sqlx"
+
+ "github.com/apache/airavata-custos/connectors/VAST/Storage-Provisioner/internal/client"
+ "github.com/apache/airavata-custos/connectors/VAST/Storage-Provisioner/internal/operations"
+ "github.com/apache/airavata-custos/connectors/VAST/Storage-Provisioner/internal/subscribers"
+ "github.com/apache/airavata-custos/internal/config"
+ "github.com/apache/airavata-custos/pkg/events"
+ "github.com/apache/airavata-custos/pkg/identity"
+ "github.com/apache/airavata-custos/pkg/service"
+)
+
+// Type is the connector type in the config file and the subscriber name on the bus.
+const Type = "vast-storage-provisioner"
+
+func LoadConnector(ctx context.Context, _ *sqlx.DB, eventBus *events.Bus, coreService *service.Service, wg *sync.WaitGroup, _ *identity.Router, connectorConfig *config.ConnectorConfig) error {
+ vms, _ := connectorConfig.GetNestedConfig("vms")
+ vmsURL, _ := vms["url"].(string)
+ username, _ := vms["username"].(string)
+ password, _ := vms["password"].(string)
+ tenantID, _ := vms["tenant_id"].(int)
+ clusterID, _ := connectorConfig.Config["custos_cluster_id"].(string)
+ storage, _ := connectorConfig.Config["storage"].([]interface{})
+ if vmsURL == "" || username == "" || password == "" || tenantID == 0 || clusterID == "" || len(storage) == 0 {
+ slog.Info("vast provisioner: required config not set; skipping")
+ return nil
+ }
+ mounts, err := parseMounts(storage)
+ if err != nil {
+ return err
+ }
+ tlsConfig, err := vmsTLS(vms)
+ if err != nil {
+ return err
+ }
+ ops := operations.New(client.New(vmsURL, username, password, tenantID, tlsConfig), coreService, clusterID, mounts)
+ subscriber := subscribers.NewStorageSubscriber(ops, eventBus, coreService, clusterID)
+ subscriber.RegisterSubscribers(Type)
+ slog.Info("vast provisioner: subscriber registered", "vms", vmsURL, "cluster_id", clusterID)
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ subscriber.StartReconciler(ctx)
+ }()
+ return nil
+}
+
+// vmsTLS trusts `ca_file` and checks the certificate against `server_name`,
+// for a VMS that still serves its self-signed default certificate.
+func vmsTLS(vms map[string]interface{}) (*tls.Config, error) {
+ caFile, _ := vms["ca_file"].(string)
+ serverName, _ := vms["server_name"].(string)
+ cfg := &tls.Config{ServerName: serverName}
+ if caFile == "" {
+ return cfg, nil
+ }
+ pem, err := os.ReadFile(caFile)
+ if err != nil {
+ return nil, err
+ }
+ cfg.RootCAs = x509.NewCertPool()
+ if !cfg.RootCAs.AppendCertsFromPEM(pem) {
+ return nil, fmt.Errorf("vast provisioner: no certificate in %s", caFile)
+ }
+ return cfg, nil
+}
+
+// parseMounts reads the `storage` list; a bad entry stops the server, since it
+// would create the wrong directories.
+func parseMounts(storage []interface{}) ([]operations.Mount, error) {
+ var mounts []operations.Mount
+ for i, e := range storage {
+ m, _ := e.(map[string]interface{})
+ str := func(k string) string { v, _ := m[k].(string); return v }
+ hardLimit, _ := m["hard_limit"].(int)
+ hardLimitInodes, _ := m["hard_limit_inodes"].(int)
+ mode, err := strconv.ParseInt(str("mode"), 8, 32)
+ mount := operations.Mount{Path: str("path"), Owner: str("owner"), Group: str("group"),
+ Mode: int(mode), HardLimit: int64(hardLimit), HardLimitInodes: int64(hardLimitInodes)}
+ switch {
+ case mount.Path == "" || mount.Owner == "" || mount.Group == "":
+ return nil, fmt.Errorf("vast provisioner: storage[%d] needs path, owner and group", i)
+ case mount.DependsOn() == "":
+ return nil, fmt.Errorf("vast provisioner: storage[%d] uses neither {user} nor {allocation}", i)
+ case err != nil:
+ return nil, fmt.Errorf("vast provisioner: storage[%d].mode must be octal, such as \"0700\": %w", i, err)
+ }
+ mounts = append(mounts, mount)
+ }
+ return mounts, nil
+}
diff --git a/connectors/VAST/Storage-Provisioner/pkg/vast/loader_test.go b/connectors/VAST/Storage-Provisioner/pkg/vast/loader_test.go
new file mode 100644
index 000000000..2f6a07197
--- /dev/null
+++ b/connectors/VAST/Storage-Provisioner/pkg/vast/loader_test.go
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package vast
+
+import (
+ "slices"
+ "testing"
+
+ "github.com/apache/airavata-custos/connectors/VAST/Storage-Provisioner/internal/operations"
+ "github.com/apache/airavata-custos/internal/config"
+)
+
+// Make sure the shipped config yields the Nexus storage policy.
+func TestShippedStorageConfig(t *testing.T) {
+ cfg, err := config.LoadConfig("../../../../../config/custos.yaml")
+ if err != nil {
+ t.Fatal(err)
+ }
+ storage, _ := cfg.Connectors["vast-storage"].Config["storage"].([]interface{})
+ got, err := parseMounts(storage)
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := []operations.Mount{
+ {Path: "/home/{user}", Owner: "{user}", Group: "{user}", Mode: 0o700, HardLimit: 100 << 30, HardLimitInodes: 1_000_000},
+ {Path: "/scratch/{user}", Owner: "{user}", Group: "{user}", Mode: 0o700, HardLimit: 10 << 40, HardLimitInodes: 10_000_000},
+ {Path: "/project/{allocation}", Owner: "0", Group: "{allocation}", Mode: 0o2750, HardLimit: 10 << 40, HardLimitInodes: 300_000_000},
+ {Path: "/project/{allocation}/shared", Owner: "0", Group: "{allocation}", Mode: 0o2770},
+ {Path: "/project/{allocation}/{user}", Owner: "{user}", Group: "{user}", Mode: 0o700},
+ }
+ if !slices.Equal(got, want) {
+ t.Fatalf("got %+v\nwant %+v", got, want)
+ }
+ for i, d := range []string{"user", "user", "allocation", "allocation", "both"} {
+ if got[i].DependsOn() != d {
+ t.Errorf("%s depends on %q, want %q", got[i].Path, got[i].DependsOn(), d)
+ }
+ }
+}
diff --git a/internal/connectors/loader.go b/internal/connectors/loader.go
index 562c0aa37..09fad4663 100644
--- a/internal/connectors/loader.go
+++ b/internal/connectors/loader.go
@@ -28,6 +28,7 @@ import (
"github.com/apache/airavata-custos/connectors/SLURM/Association-Mapper/pkg/smapper"
"github.com/apache/airavata-custos/connectors/SLURM/Usage-Monitor/pkg/monitor"
"github.com/apache/airavata-custos/connectors/TempAccount/pkg/tempaccount"
+ "github.com/apache/airavata-custos/connectors/VAST/Storage-Provisioner/pkg/vast"
"github.com/apache/airavata-custos/internal/config"
"github.com/apache/airavata-custos/pkg/events"
"github.com/apache/airavata-custos/pkg/identity"
@@ -45,6 +46,7 @@ func LoadConnectorsFromConfig(ctx context.Context, cfg *config.Config, database
monitor.Type: monitor.LoadConnector,
tempaccount.Type: tempaccount.LoadConnector,
analytics.Type: analytics.LoadConnector,
+ vast.Type: vast.LoadConnector,
}
for connectorName, connectorCfg := range cfg.Connectors {