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/10/12 10:33:59 UTC
(airavata-custos) branch master updated: Reactivate accounts and projects on repeat AMIE create packets (#628)
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 97e86a965 Reactivate accounts and projects on repeat AMIE create packets (#628)
97e86a965 is described below
commit 97e86a9659f0c9b9725ae286fde288e0cb897ee3
Author: Lahiru Jayathilake <la...@gmail.com>
AuthorDate: Mon Oct 12 06:33:53 2026 -0400
Reactivate accounts and projects on repeat AMIE create packets (#628)
* Reactivate accounts and projects on repeat AMIE create packets
* Show failed AMIE packets on the trace and reject packets without an email
* Give the AMIE failure trace rows their own span
---
.../ACCESS/AMIE-Processor/handler/handler.go | 9 ++++
.../handler/request_account_create.go | 21 +++++---
.../request_account_create_integration_test.go | 34 +++++++++++++
.../handler/request_project_create.go | 12 ++++-
.../request_project_create_integration_test.go | 59 ++++++++++++++++++++++
connectors/ACCESS/AMIE-Processor/model/audit.go | 1 +
.../AMIE-Processor/pipeline/integration_common.go | 2 +-
.../ACCESS/AMIE-Processor/pkg/amie/loader.go | 2 +-
.../ACCESS/AMIE-Processor/worker/processor.go | 28 +++++++++-
.../worker/processor_failures_integration_test.go | 35 +++++++++++++
.../worker/processor_integration_test.go | 11 ++++
11 files changed, 201 insertions(+), 13 deletions(-)
diff --git a/connectors/ACCESS/AMIE-Processor/handler/handler.go b/connectors/ACCESS/AMIE-Processor/handler/handler.go
index 1bc1639f0..f3cb91bea 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/handler.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/handler.go
@@ -135,6 +135,15 @@ func ensureUserIdentity(ctx context.Context, svc *service.Service, userID, globa
return nil
}
+// reactivate sets an existing membership back to active. The spec says a
+// create packet for a known account also reactivates it.
+func reactivate(ctx context.Context, svc *service.Service, m models.ComputeAllocationMembership) (*models.ComputeAllocationMembership, error) {
+ if m.MembershipStatus == models.ACTIVE {
+ return &m, nil
+ }
+ return svc.UpdateMembershipStatus(ctx, m.ID, models.ACTIVE)
+}
+
// flipUserMemberships flips every ComputeAllocationMembership the user holds
// under any allocation belonging to the given project (Custos project.id) to
// the given status. Returns the rows that were updated. Silently returns an
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_account_create.go b/connectors/ACCESS/AMIE-Processor/handler/request_account_create.go
index 2f479883c..56f1d2672 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_account_create.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_account_create.go
@@ -74,6 +74,9 @@ func (h *RequestAccountCreateHandler) Handle(ctx context.Context, tx *sql.Tx, pa
if err := requireText(userGlobalID, "UserGlobalID"); err != nil {
return err
}
+ if err := requireText(getString(body, "UserEmail"), "UserEmail"); err != nil {
+ return err
+ }
if h.clusterID == "" {
return fmt.Errorf("AMIE_CLUSTER_ID not configured")
}
@@ -111,7 +114,7 @@ func (h *RequestAccountCreateHandler) Handle(ctx context.Context, tx *sql.Tx, pa
}
role := normalizeAMIERole(normalizeRole(getString(body, "UserRole")))
- membership, err := h.ensureMembership(ctx, allocation.ID, user.ID)
+ membership, err := h.ensureMembership(ctx, &allocation, user.ID)
if err != nil {
return fmt.Errorf("request_account_create: ensure membership: %w", err)
}
@@ -188,21 +191,23 @@ func (h *RequestAccountCreateHandler) ensureComputeClusterUser(ctx context.Conte
return allocateAndCreateClusterUser(ctx, h.svc, h.clusterID, userID)
}
-// ensureMembership returns the existing (allocation, user) membership or
-// creates a new one. Idempotent for re-delivered packets.
-func (h *RequestAccountCreateHandler) ensureMembership(ctx context.Context, allocationID, userID string) (*models.ComputeAllocationMembership, error) {
- existing, err := h.svc.ListMembersForAllocation(ctx, allocationID)
+// ensureMembership reactivates a known member only while the allocation is active.
+// On an inactive project the member stays inactive, as the spec says.
+func (h *RequestAccountCreateHandler) ensureMembership(ctx context.Context, allocation *models.ComputeAllocation, userID string) (*models.ComputeAllocationMembership, error) {
+ existing, err := h.svc.ListMembersForAllocation(ctx, allocation.ID)
if err != nil {
return nil, fmt.Errorf("list memberships: %w", err)
}
for _, m := range existing {
if m.UserID == userID {
- cm := m.ComputeAllocationMembership
- return &cm, nil
+ if allocation.Status != models.ACTIVE {
+ return &m.ComputeAllocationMembership, nil
+ }
+ return reactivate(ctx, h.svc, m.ComputeAllocationMembership)
}
}
return h.svc.CreateComputeAllocationMembership(ctx, &models.ComputeAllocationMembership{
- ComputeAllocationID: allocationID,
+ ComputeAllocationID: allocation.ID,
UserID: userID,
})
}
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_account_create_integration_test.go b/connectors/ACCESS/AMIE-Processor/handler/request_account_create_integration_test.go
index 8f8a318bd..e060ed448 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_account_create_integration_test.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_account_create_integration_test.go
@@ -365,6 +365,40 @@ func TestRequestAccountCreate_ApprovedAccountRepliesAtOnce(t *testing.T) {
}
}
+// Make sure an account request for an inactivated member of an active
+// allocation sets the membership back to active.
+func TestRequestAccountCreate_ReactivatesInactiveMember(t *testing.T) {
+ database := setupTestDB(t)
+ projectID := seedProjectForRAC(t, database)
+ body := baseRACBody()
+ body["ProjectID"] = projectID
+
+ svc := newTestCoreService(database)
+ h := NewRequestAccountCreateHandler(svc, testClusterID, &fakeAmieClient{}, newTestAuditService(database))
+ handle := func() {
+ pkt := insertPacket(t, database, "request_account_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: %v", err)
+ }
+ }
+ handle()
+ const member = "user_id = (SELECT id FROM users WHERE email = $1)"
+ if _, err := database.Exec("UPDATE compute_allocation_memberships SET membership_status = 'INACTIVE' WHERE "+member, body["UserEmail"]); err != nil {
+ t.Fatalf("inactivate membership: %v", err)
+ }
+ handle()
+
+ var status string
+ if err := database.Get(&status, "SELECT membership_status FROM compute_allocation_memberships WHERE "+member, body["UserEmail"]); err != nil {
+ t.Fatalf("read membership: %v", err)
+ }
+ if status != "ACTIVE" {
+ t.Errorf("membership_status: got %s, want ACTIVE", status)
+ }
+}
+
// Make sure an account request for the project's PI, under any role, adds the
// membership and leaves the PI and project roles untouched.
func TestRequestAccountCreate_ForThePI(t *testing.T) {
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_project_create.go b/connectors/ACCESS/AMIE-Processor/handler/request_project_create.go
index e05bacb61..c8f7f321a 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_project_create.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_project_create.go
@@ -75,6 +75,9 @@ func (h *RequestProjectCreateHandler) Handle(ctx context.Context, tx *sql.Tx, pa
if err := requireText(piGlobalID, "PiGlobalID"); err != nil {
return err
}
+ if err := requireText(getString(body, "PiEmail"), "PiEmail"); err != nil {
+ return err
+ }
if h.clusterID == "" {
return fmt.Errorf("AMIE_CLUSTER_ID not configured")
}
@@ -207,6 +210,10 @@ func (h *RequestProjectCreateHandler) ensurePIClusterUser(ctx context.Context, u
func (h *RequestProjectCreateHandler) ensureProject(ctx context.Context, originatedID, title, piID string) (*models.Project, error) {
if p, err := h.svc.GetProjectByOriginatedID(ctx, originatedID); err == nil {
+ // The spec says a project packet for an inactive project reactivates it.
+ if p.Status != models.ProjectActive {
+ return h.svc.UpdateProjectStatus(ctx, p.ID, models.ProjectActive)
+ }
return p, nil
} else if !errors.Is(err, service.ErrNotFound) {
return nil, err
@@ -299,8 +306,8 @@ func (h *RequestProjectCreateHandler) ensurePIMembership(ctx context.Context, al
}
for _, m := range existing {
if m.UserID == piID {
- cm := m.ComputeAllocationMembership
- return &cm, false, nil
+ active, err := reactivate(ctx, h.svc, m.ComputeAllocationMembership)
+ return active, false, err
}
}
created, err := h.svc.CreateComputeAllocationMembership(ctx, &models.ComputeAllocationMembership{
@@ -344,6 +351,7 @@ func (h *RequestProjectCreateHandler) changeAllocation(ctx context.Context, body
// NEW, TRANSFER, SUPPLEMENT, ADJUSTMENT, ADVANCE. The amount is signed, so a negative transfer deducts.
alloc.InitialSUAmount += su
}
+ alloc.Status = models.ACTIVE
if err := h.svc.UpdateComputeAllocation(ctx, alloc); err != nil {
return nil, fmt.Errorf("change allocation: %w", err)
}
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 5d4cf743d..344811716 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
@@ -388,6 +388,46 @@ func TestRequestProjectCreate_ExtensionMovesEndDate(t *testing.T) {
}
}
+// Make sure a project packet for an inactivated project sets the project, its
+// allocation and the PI's membership back to active.
+func TestRequestProjectCreate_ReactivatesInactiveProject(t *testing.T) {
+ database := setupTestDB(t)
+ seedProjectForRAC(t, database)
+ for _, q := range []string{
+ "UPDATE projects SET status = 'INACTIVE'",
+ "UPDATE compute_allocations SET status = 'INACTIVE'",
+ "UPDATE compute_allocation_memberships SET membership_status = 'INACTIVE'",
+ } {
+ if _, err := database.Exec(q); err != nil {
+ t.Fatalf("%s: %v", q, err)
+ }
+ }
+
+ svc := newTestCoreService(database)
+ h := NewRequestProjectCreateHandler(svc, testClusterID, &fakeAmieClient{}, newTestAuditService(database), nil)
+ 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: %v", err)
+ }
+
+ var statuses []string
+ if err := database.Select(&statuses,
+ `SELECT status FROM projects
+ UNION ALL SELECT status FROM compute_allocations
+ UNION ALL SELECT membership_status FROM compute_allocation_memberships`); err != nil {
+ t.Fatalf("read statuses: %v", err)
+ }
+ for _, s := range statuses {
+ if s != "ACTIVE" {
+ t.Errorf("statuses: got %v, want all ACTIVE", statuses)
+ break
+ }
+ }
+}
+
// Make sure a negative transfer deducts its exact amount, fraction included,
// from the existing allocation.
func TestRequestProjectCreate_NegativeTransferDeducts(t *testing.T) {
@@ -402,6 +442,25 @@ func TestRequestProjectCreate_NegativeTransferDeducts(t *testing.T) {
}
}
+// Make sure a project packet whose PI has no email is rejected as invalid and creates nothing.
+func TestRequestProjectCreate_MissingPIEmailIsInvalid(t *testing.T) {
+ database := setupTestDB(t)
+ body := baseRPCBody()
+ delete(body, "PiEmail")
+ pkt := insertPacket(t, database, "request_project_create", body)
+ h := NewRequestProjectCreateHandler(newTestCoreService(database), testClusterID, &fakeAmieClient{}, newTestAuditService(database), nil)
+
+ 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, "")
+ })
+ if !errors.Is(err, model.ErrInvalidPacket) {
+ t.Fatalf("error: got %v, want an invalid packet error", err)
+ }
+ if got := countRows(t, database, "organizations"); got != 0 {
+ t.Errorf("organizations: got %d, want 0", got)
+ }
+}
+
// TestRequestProjectCreate_MissingGrantNumber asserts the handler rejects a
// packet missing a the protocol-required field, without mutating the DB.
func TestRequestProjectCreate_MissingGrantNumber(t *testing.T) {
diff --git a/connectors/ACCESS/AMIE-Processor/model/audit.go b/connectors/ACCESS/AMIE-Processor/model/audit.go
index 8b6d54182..b248c39c4 100644
--- a/connectors/ACCESS/AMIE-Processor/model/audit.go
+++ b/connectors/ACCESS/AMIE-Processor/model/audit.go
@@ -22,6 +22,7 @@ type AuditAction string
const (
AuditPacketReceived AuditAction = "PACKET_RECEIVED"
+ AuditPacketFailed AuditAction = "PACKET_FAILED"
AuditCreatePerson AuditAction = "CREATE_PERSON"
AuditUpdatePerson AuditAction = "UPDATE_PERSON"
AuditDeletePerson AuditAction = "DELETE_PERSON"
diff --git a/connectors/ACCESS/AMIE-Processor/pipeline/integration_common.go b/connectors/ACCESS/AMIE-Processor/pipeline/integration_common.go
index 7ee642d3b..28476304f 100644
--- a/connectors/ACCESS/AMIE-Processor/pipeline/integration_common.go
+++ b/connectors/ACCESS/AMIE-Processor/pipeline/integration_common.go
@@ -217,7 +217,7 @@ func newTestPipeline(t *testing.T) *testPipeline {
met := metrics.NewWithRegistry(prometheus.NewRegistry())
poller := worker.NewPoller(amieClient, packetStore, eventStore, met, database, cfg)
- processor := worker.NewProcessor(eventStore, packetStore, errorStore, router, met, auditSvc, database, cfg)
+ processor := worker.NewProcessor(eventStore, packetStore, errorStore, router, met, auditSvc, amieClient, database, cfg)
ctx, cancel := context.WithCancel(context.Background())
pipe := &testPipeline{
diff --git a/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go b/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go
index 628089068..4c07bb5ef 100644
--- a/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go
+++ b/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go
@@ -116,7 +116,7 @@ func LoadConnector(ctx context.Context, database *sqlx.DB, eventBus *events.Bus,
met := metrics.New()
poller := worker.NewPoller(amie, packetStore, eventStore, met, database, cfg)
- processor := worker.NewProcessor(eventStore, packetStore, errorStore, dispatcher, met, auditSvc, database, cfg)
+ processor := worker.NewProcessor(eventStore, packetStore, errorStore, dispatcher, met, auditSvc, amie, database, cfg)
wg.Add(2)
go func() {
diff --git a/connectors/ACCESS/AMIE-Processor/worker/processor.go b/connectors/ACCESS/AMIE-Processor/worker/processor.go
index 4b9d36d6e..4f715b5a5 100644
--- a/connectors/ACCESS/AMIE-Processor/worker/processor.go
+++ b/connectors/ACCESS/AMIE-Processor/worker/processor.go
@@ -81,6 +81,10 @@ type processorAuditService interface {
Log(ctx context.Context, tx *sql.Tx, packetID, eventID string, action model.AuditAction, entityType, entityID, summary string) error
}
+type processorReplier interface {
+ ReplyToPacket(ctx context.Context, packetRecID int64, reply map[string]any) error
+}
+
type Processor struct {
eventStore processorEventStore
packetStore processorPacketStore
@@ -88,11 +92,12 @@ type Processor struct {
router processorRouter
metrics processorMetrics
auditSvc processorAuditService
+ amie processorReplier
db *sqlx.DB
workerInterval time.Duration
}
-func NewProcessor(eventStore processorEventStore, packetStore processorPacketStore, errorStore processorErrorStore, router processorRouter, metrics processorMetrics, auditSvc processorAuditService, db *sqlx.DB, cfg amieclient.Config) *Processor {
+func NewProcessor(eventStore processorEventStore, packetStore processorPacketStore, errorStore processorErrorStore, router processorRouter, metrics processorMetrics, auditSvc processorAuditService, amie processorReplier, db *sqlx.DB, cfg amieclient.Config) *Processor {
return &Processor{
eventStore: eventStore,
packetStore: packetStore,
@@ -100,6 +105,7 @@ func NewProcessor(eventStore processorEventStore, packetStore processorPacketSto
router: router,
metrics: metrics,
auditSvc: auditSvc,
+ amie: amie,
db: db,
workerInterval: cfg.WorkerInterval,
}
@@ -251,6 +257,9 @@ func (p *Processor) executeInTransaction(ctx context.Context, ewp model.EventWit
// the event afresh (with the pre-rollback attempt count) and schedules a retry
// or marks the event as permanently failed.
func (p *Processor) recordFailureInNewTransaction(ctx context.Context, eventID string, cause error) error {
+ // Its own span, so the audit rows written here have a parent on the trace.
+ ctx, span := tracing.Start(ctx, "amie.record_failure")
+ defer span.End()
return custosdb.TxFn(ctx, p.db, func(tx *sql.Tx) error {
// Re-read the event -- the processing transaction was rolled back,
// so the attempt count is still the pre-increment value.
@@ -309,6 +318,23 @@ func (p *Processor) recordFailureInNewTransaction(ctx context.Context, eventID s
if err := p.packetStore.Update(ctx, tx, packet); err != nil {
return fmt.Errorf("update packet status: %w", err)
}
+ // The handler's audit rows rolled back with it. Write them here so the failed packet shows up on the trace.
+ if err := p.auditSvc.Log(ctx, tx, packet.ID, event.ID, model.AuditPacketReceived, "packet", packet.ID, packet.Type); err != nil {
+ return fmt.Errorf("audit PACKET_RECEIVED: %w", err)
+ }
+ if err := p.auditSvc.Log(ctx, tx, packet.ID, event.ID, model.AuditPacketFailed, "packet", packet.ID, errMsg); err != nil {
+ return fmt.Errorf("audit PACKET_FAILED: %w", err)
+ }
+ // An invalid packet never succeeds, so end its transaction at ACCESS. Other failures wait for an admin to fix and retry.
+ if errors.Is(cause, model.ErrInvalidPacket) {
+ failure := map[string]any{
+ "type": "inform_transaction_complete",
+ "body": map[string]any{"StatusCode": "Failure", "DetailCode": 2, "Message": errMsg},
+ }
+ if err := p.amie.ReplyToPacket(ctx, packet.AmieID, failure); err != nil {
+ slog.ErrorContext(ctx, "failure reply not sent", "packetId", packet.ID, "error", err)
+ }
+ }
}
if err := p.eventStore.Update(ctx, tx, event); err != nil {
diff --git a/connectors/ACCESS/AMIE-Processor/worker/processor_failures_integration_test.go b/connectors/ACCESS/AMIE-Processor/worker/processor_failures_integration_test.go
index b5e68c303..33f92955f 100644
--- a/connectors/ACCESS/AMIE-Processor/worker/processor_failures_integration_test.go
+++ b/connectors/ACCESS/AMIE-Processor/worker/processor_failures_integration_test.go
@@ -23,9 +23,12 @@ import (
"context"
"database/sql"
"errors"
+ "fmt"
+ "strings"
"testing"
"github.com/apache/airavata-custos/connectors/ACCESS/AMIE-Processor/model"
+ corestore "github.com/apache/airavata-custos/internal/store"
"github.com/google/uuid"
)
@@ -118,6 +121,38 @@ func TestProcessor_FAILEDStatusIsUnreachable(t *testing.T) {
}
}
+// Make sure an invalid packet fails at once, is recorded as failed on its trace, and gets one failure reply.
+func TestProcessor_InvalidPacketFailsWithTraceAndReply(t *testing.T) {
+ database := setupTestDB(t)
+ packetID, eventID := seedNewEvent(t, database)
+ replier := &stubReplier{}
+ p := newProcessor(database, &fakeRouter{errors: []error{
+ fmt.Errorf("%w: 'PiEmail' must not be empty", model.ErrInvalidPacket),
+ }}, &stubMetrics{})
+ p.amie = replier
+
+ p.processPendingEvents(context.Background())
+
+ if ev := readEvent(t, database, eventID); ev.Status != model.ProcessingStatusPermanentlyFailed {
+ t.Errorf("event status: got %q, want PERMANENTLY_FAILED", ev.Status)
+ }
+ var actions []string
+ if err := database.Select(&actions,
+ "SELECT event_type FROM audit_events WHERE entity_id = $1 ORDER BY event_time", packetID); err != nil {
+ t.Fatalf("read audit: %v", err)
+ }
+ if strings.Join(actions, ",") != "PACKET_RECEIVED,PACKET_FAILED" {
+ t.Errorf("audit rows: got %v, want PACKET_RECEIVED then PACKET_FAILED", actions)
+ }
+ tree, _, _, err := corestore.NewAuditTraceStore(database).GetTraceTree(context.Background(), pollTraceID)
+ if err != nil || tree == nil || len(tree.Children) == 0 {
+ t.Errorf("trace tree: got %+v (%v), want the failed packet's steps", tree, err)
+ }
+ if len(replier.replies) != 1 || replier.replies[0]["body"].(map[string]any)["StatusCode"] != "Failure" {
+ t.Errorf("replies: got %v, want one Failure reply", replier.replies)
+ }
+}
+
// probeRouter does a configurable write inside the handler's tx, then
// returns a configurable error. Lets tests verify tx atomicity.
type probeRouter struct {
diff --git a/connectors/ACCESS/AMIE-Processor/worker/processor_integration_test.go b/connectors/ACCESS/AMIE-Processor/worker/processor_integration_test.go
index 59a65fa10..718abafa3 100644
--- a/connectors/ACCESS/AMIE-Processor/worker/processor_integration_test.go
+++ b/connectors/ACCESS/AMIE-Processor/worker/processor_integration_test.go
@@ -97,11 +97,22 @@ func newProcessor(database *sqlx.DB, router processorRouter, met *stubMetrics) *
router,
met,
service.NewAuditService(corestore.NewAuditEventStore(database), store.NewAuditExtrasStore(database)),
+ &stubReplier{},
database,
cfg,
)
}
+// stubReplier keeps the replies the processor sends to AMIE.
+type stubReplier struct {
+ replies []map[string]any
+}
+
+func (s *stubReplier) ReplyToPacket(_ context.Context, _ int64, reply map[string]any) error {
+ s.replies = append(s.replies, reply)
+ return nil
+}
+
func readEvent(t *testing.T, database *sqlx.DB, id string) *model.ProcessingEvent {
t.Helper()
var e model.ProcessingEvent