You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by la...@apache.org on 2026/07/21 08:56:30 UTC
(airavata-custos) branch master updated: Persist the resource layer for AMIE-ingested allocations (#546)
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake 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 3cf89c227 Persist the resource layer for AMIE-ingested allocations (#546)
3cf89c227 is described below
commit 3cf89c227065bf151876532e9aa06918cac89f12
Author: Lahiru Jayathilake <la...@gmail.com>
AuthorDate: Tue Jul 21 04:56:24 2026 -0400
Persist the resource layer for AMIE-ingested allocations (#546)
* Create resource mappings from AMIE project packets
Fixes #539
* Cover AMIE resource mapping creation in the baseline scenario
---
.../handler/request_project_create.go | 34 +++++++++++
.../request_project_create_integration_test.go | 71 ++++++++++++++++++++++
connectors/ACCESS/AMIE-Processor/model/audit.go | 1 +
.../pipeline/baseline_integration_test.go | 8 ++-
.../AMIE-Processor/pipeline/integration_common.go | 8 +++
.../testdata/scenarios/baseline.yaml | 14 ++++-
6 files changed, 132 insertions(+), 4 deletions(-)
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_project_create.go b/connectors/ACCESS/AMIE-Processor/handler/request_project_create.go
index 1574293bf..37edce93c 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_project_create.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_project_create.go
@@ -22,6 +22,7 @@ import (
"database/sql"
"errors"
"fmt"
+ "log/slog"
"strings"
"go.opentelemetry.io/otel/codes"
@@ -114,6 +115,10 @@ func (h *RequestProjectCreateHandler) Handle(ctx context.Context, tx *sql.Tx, pa
return fmt.Errorf("request_project_create: audit CREATE_ALLOCATION: %w", err)
}
+ if err := h.ensureResourceMappings(ctx, tx, body, allocation, packet, eventID); err != nil {
+ return fmt.Errorf("request_project_create: ensure resource mappings: %w", err)
+ }
+
piMembership, created, err := h.ensurePIMembership(ctx, allocation, pi.ID)
if err != nil {
return fmt.Errorf("request_project_create: ensure PI membership: %w", err)
@@ -240,6 +245,35 @@ func (h *RequestProjectCreateHandler) ensureAllocation(ctx context.Context, body
})
}
+// ensureResourceMappings grants the allocation each packet resource found in
+// the cluster's catalog. AMIE expresses the grant only in service units,
+// already stored on the allocation, so mappings carry no native caps;
+// unregistered names are skipped rather than failing the packet.
+func (h *RequestProjectCreateHandler) ensureResourceMappings(ctx context.Context, tx *sql.Tx, body map[string]any, allocation *models.ComputeAllocation, packet *model.Packet, eventID string) error {
+ for _, name := range getResourceList(body) {
+ resource, err := h.svc.GetComputeAllocationResourceByNameAndCluster(ctx, name, h.clusterID)
+ if errors.Is(err, service.ErrNotFound) {
+ slog.Warn("amie: packet resource not in the cluster catalog, mapping skipped",
+ "resource", name, "allocation_id", allocation.ID)
+ continue
+ }
+ if err != nil {
+ return fmt.Errorf("lookup resource %q: %w", name, err)
+ }
+ mapping, err := h.svc.AttachResourceToAllocation(ctx, allocation.ID, resource.ID, 0, 0)
+ if errors.Is(err, service.ErrAlreadyExists) {
+ continue
+ }
+ if err != nil {
+ return fmt.Errorf("attach resource %q: %w", name, err)
+ }
+ if err := h.auditSvc.Log(ctx, tx, packet.ID, eventID, model.AuditAttachResource, "compute_allocation_resource_mapping", mapping.ID, name); err != nil {
+ return fmt.Errorf("audit ATTACH_RESOURCE: %w", err)
+ }
+ }
+ return nil
+}
+
// ensurePIMembership asserts the PI as a role=PI membership on the allocation.
// Returns (membership, created=true) when it inserted a new row, or
// (existing, created=false) on redelivery. The service rejects a second
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_project_create_integration_test.go b/connectors/ACCESS/AMIE-Processor/handler/request_project_create_integration_test.go
index 7400715f5..f281fc936 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_project_create_integration_test.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_project_create_integration_test.go
@@ -27,6 +27,7 @@ import (
"testing"
"github.com/apache/airavata-custos/connectors/ACCESS/AMIE-Processor/model"
+ "github.com/apache/airavata-custos/pkg/models"
)
// baseRPCBody returns a fully populated request_project_create body with
@@ -91,6 +92,10 @@ func TestRequestProjectCreate_HappyPath(t *testing.T) {
if got := countRows(t, database, "compute_allocations"); got != 1 {
t.Errorf("compute_allocations: got %d, want 1", got)
}
+ // Unregistered packet resources create no mapping and do not fail the packet.
+ if got := countRows(t, database, "compute_allocation_resource_mappings"); got != 0 {
+ t.Errorf("compute_allocation_resource_mappings: got %d, want 0 (resource not in catalog)", got)
+ }
var org struct {
OriginatedID string `db:"originated_id"`
@@ -356,5 +361,71 @@ func TestRequestProjectCreate_ReplyFailurePropagates(t *testing.T) {
}
}
+// TestRequestProjectCreate_AttachesRegisteredResources asserts a catalog-registered
+// packet resource is mapped with no caps and not duplicated on re-delivery.
+func TestRequestProjectCreate_AttachesRegisteredResources(t *testing.T) {
+ database := setupTestDB(t)
+
+ svc := newTestCoreService(database)
+ audit := newTestAuditService(database)
+ amie := &fakeAmieClient{}
+ h := NewRequestProjectCreateHandler(svc, testClusterID, amie, audit)
+
+ resource, err := svc.CreateComputeAllocationResource(context.Background(), &models.ComputeAllocationResource{
+ Name: "compute.access-ci.org",
+ ResourceType: "CPU_HOURS",
+ ResourceAmount: 1000,
+ ComputeClusterID: testClusterID,
+ })
+ if err != nil {
+ t.Fatalf("seed catalog resource: %v", err)
+ }
+
+ body := baseRPCBody()
+ pkt := insertPacket(t, database, "request_project_create", body)
+ if err := runHandlerInTx(t, database, func(ctx context.Context, tx *sql.Tx) error {
+ return h.Handle(ctx, tx, map[string]any{"type": pkt.Type, "body": body}, pkt, "")
+ }); err != nil {
+ t.Fatalf("Handle returned error: %v", err)
+ }
+
+ var mapping struct {
+ ComputeAllocationResourceID string `db:"compute_allocation_resource_id"`
+ ResourceAmount int64 `db:"resource_amount"`
+ ResourceTime int64 `db:"resource_time"`
+ }
+ if err := database.Get(&mapping,
+ "SELECT compute_allocation_resource_id, resource_amount, resource_time FROM compute_allocation_resource_mappings LIMIT 1",
+ ); err != nil {
+ t.Fatalf("read mapping: %v", err)
+ }
+ if mapping.ComputeAllocationResourceID != resource.ID {
+ t.Errorf("mapping.resource_id: got %q, want %q", mapping.ComputeAllocationResourceID, resource.ID)
+ }
+ // The packet carries no native amounts; the SU budget is the limit.
+ if mapping.ResourceAmount != 0 || mapping.ResourceTime != 0 {
+ t.Errorf("mapping caps: got (%d,%d), want (0,0)", mapping.ResourceAmount, mapping.ResourceTime)
+ }
+ if got := countAuditActions(t, database, pkt.ID, model.AuditAttachResource); got != 1 {
+ t.Errorf("audit ATTACH_RESOURCE: got %d, want 1", got)
+ }
+
+ // Re-delivery (supplement) must not duplicate the mapping.
+ second := baseRPCBody()
+ second["AllocationType"] = "supplement"
+ secondPkt := insertPacket(t, database, "request_project_create", second)
+ if err := runHandlerInTx(t, database, func(ctx context.Context, tx *sql.Tx) error {
+ return h.Handle(ctx, tx, map[string]any{"type": secondPkt.Type, "body": second}, secondPkt, "")
+ }); err != nil {
+ t.Fatalf("second Handle: %v", err)
+ }
+ if got := countRows(t, database, "compute_allocation_resource_mappings"); got != 1 {
+ t.Errorf("mappings after re-delivery: got %d, want 1", got)
+ }
+ if got := countAuditActions(t, database, secondPkt.ID, model.AuditAttachResource); got != 0 {
+ t.Errorf("audit ATTACH_RESOURCE on re-delivery: got %d, want 0", got)
+ }
+}
+
// silence unused-import false positives when only some tests reference these.
var _ = model.AuditCreatePerson
diff --git a/connectors/ACCESS/AMIE-Processor/model/audit.go b/connectors/ACCESS/AMIE-Processor/model/audit.go
index 6938492b2..778338aa9 100644
--- a/connectors/ACCESS/AMIE-Processor/model/audit.go
+++ b/connectors/ACCESS/AMIE-Processor/model/audit.go
@@ -29,6 +29,7 @@ const (
AuditCreateAccount AuditAction = "CREATE_ACCOUNT"
AuditCreateProject AuditAction = "CREATE_PROJECT"
AuditCreateAllocation AuditAction = "CREATE_ALLOCATION"
+ AuditAttachResource AuditAction = "ATTACH_RESOURCE"
AuditInactivateProject AuditAction = "INACTIVATE_PROJECT"
AuditReactivateProject AuditAction = "REACTIVATE_PROJECT"
AuditCreateMembership AuditAction = "CREATE_MEMBERSHIP"
diff --git a/connectors/ACCESS/AMIE-Processor/pipeline/baseline_integration_test.go b/connectors/ACCESS/AMIE-Processor/pipeline/baseline_integration_test.go
index f1b439eec..5f5d57a18 100644
--- a/connectors/ACCESS/AMIE-Processor/pipeline/baseline_integration_test.go
+++ b/connectors/ACCESS/AMIE-Processor/pipeline/baseline_integration_test.go
@@ -48,10 +48,12 @@ func TestPipeline_BaselineDeterminism(t *testing.T) {
{"compute_allocations", 2},
{"compute_allocation_diffs", 1},
{"amie_user_dns", 2},
- {"amie_audit_extras", 52},
+ {"amie_audit_extras", 54},
{"compute_cluster_users", 4},
{"compute_allocation_memberships", 5},
{"project_memberships", 4},
+ // One per allocation; the supplement re-delivery does not duplicate.
+ {"compute_allocation_resource_mappings", 2},
}
if decoded != 12 {
t.Errorf("decoded packets: got %d, want 12", decoded)
@@ -69,8 +71,8 @@ func TestPipeline_BaselineDeterminism(t *testing.T) {
); err != nil {
t.Fatalf("count amie audit_events: %v", err)
}
- if amieAuditEvents != 52 {
- t.Errorf("audit_events source='amie': got %d, want 52", amieAuditEvents)
+ if amieAuditEvents != 54 {
+ t.Errorf("audit_events source='amie': got %d, want 54", amieAuditEvents)
}
// audit_log.by_action.
diff --git a/connectors/ACCESS/AMIE-Processor/pipeline/integration_common.go b/connectors/ACCESS/AMIE-Processor/pipeline/integration_common.go
index 6d58e69fd..cd6d2383c 100644
--- a/connectors/ACCESS/AMIE-Processor/pipeline/integration_common.go
+++ b/connectors/ACCESS/AMIE-Processor/pipeline/integration_common.go
@@ -164,6 +164,14 @@ func seedCluster(t *testing.T, database *sqlx.DB) {
); err != nil {
t.Fatalf("seed cluster: %v", err)
}
+ // Catalog entry matching the mock packets' ResourceList, so
+ // request_project_create exercises the resource-mapping path.
+ if _, err := database.Exec(
+ "INSERT INTO compute_allocation_resources (id, name, resource_type, resource_amount, compute_cluster_id) VALUES (?, ?, ?, ?, ?)",
+ "baseline-resource", "baseline-cluster.example.edu", "CPU_HOURS", 0, testClusterID,
+ ); err != nil {
+ t.Fatalf("seed catalog resource: %v", err)
+ }
}
// testPipeline runs the AMIE connector in-process against the local mock
diff --git a/connectors/ACCESS/AMIE-Processor/testdata/scenarios/baseline.yaml b/connectors/ACCESS/AMIE-Processor/testdata/scenarios/baseline.yaml
index 9c8df9dd6..abee44cda 100644
--- a/connectors/ACCESS/AMIE-Processor/testdata/scenarios/baseline.yaml
+++ b/connectors/ACCESS/AMIE-Processor/testdata/scenarios/baseline.yaml
@@ -46,6 +46,9 @@ trigger:
seed:
- table: compute_clusters
row: { id: "${AMIE_CLUSTER_ID}", name: default-cluster }
+ - table: compute_allocation_resources
+ # Matches every packet's ResourceList so project_create attaches it.
+ row: { id: baseline-resource, name: baseline-cluster.example.edu, resource_type: CPU_HOURS, resource_amount: 0, compute_cluster_id: "${AMIE_CLUSTER_ID}" }
expectations:
@@ -155,16 +158,25 @@ expectations:
CO_PI: 1
ALLOCATION_MANAGER: 1
+ compute_allocation_resource_mappings:
+ # One mapping per allocation from the seeded catalog row; the
+ # supplement re-delivery does not duplicate.
+ total_count: 2
+ rows:
+ - { compute_allocation_resource_id: baseline-resource, resource_amount: 0, resource_time: 0 }
+ - { compute_allocation_resource_id: baseline-resource, resource_amount: 0, resource_time: 0 }
+
audit_log:
# audit_events (source='amie') and amie_audit_extras carry the same count;
# one row each per AMIE audit.
- total_count: 52
+ total_count: 54
by_action:
PACKET_RECEIVED: 12 # one per packet
CREATE_PERSON: 6 # 3 project_create + 3 account_create
CREATE_ACCOUNT: 6 # 3 project_create + 3 account_create
CREATE_PROJECT: 3
CREATE_ALLOCATION: 3 # supplement-delivery still audits CREATE_ALLOCATION
+ ATTACH_RESOURCE: 2 # one per allocation; supplement attach is a no-op
CREATE_MEMBERSHIP: 5 # 2 PIs + 3 account_create
REPLY_SENT: 11 # everything except inform_transaction_complete
PERSIST_DNS: 3 # 2 from data_*, 1 delete from request_user_modify