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

Author: lahirujayathilake

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

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

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"

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.

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

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