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/09/25 16:03:11 UTC
(airavata-custos) 02/05: Return errors from event subscribers and register them by name
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake pushed a commit to branch persistent-event-bus
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit 6bb4f5e3d6261b860d53ce54222804d95b6908dd
Author: lahiruj <la...@gmail.com>
AuthorDate: Tue Sep 22 15:56:45 2026 -0400
Return errors from event subscribers and register them by name
---
.../ACCESS/AMIE-Processor/pkg/amie/loader.go | 3 +
connectors/Analytics/pkg/analytics/loader.go | 3 +
.../internal/subscribers/cluster_user.go | 25 +--
.../Identity-Provisioner/pkg/comanage/loader.go | 5 +-
.../internal/subscribers/account.go | 42 ++---
.../internal/subscribers/association_writer.go | 3 +
.../internal/subscribers/members.go | 35 ++--
.../internal/subscribers/members_test.go | 30 +++-
.../internal/subscribers/subscriber.go | 19 ++-
.../SLURM/Association-Mapper/pkg/smapper/loader.go | 5 +-
.../SLURM/Usage-Monitor/pkg/monitor/loader.go | 3 +
connectors/TempAccount/pkg/tempaccount/loader.go | 3 +
internal/connectors/loader.go | 12 +-
pkg/events/bus.go | 70 ++++++--
pkg/events/bus_test.go | 188 ++++++++++-----------
pkg/events/compute_allocation_diff_subscribe.go | 33 +---
...ation_membership_resource_override_subscribe.go | 33 +---
.../compute_allocation_membership_subscribe.go | 33 +---
...ompute_allocation_resource_mapping_subscribe.go | 33 +---
.../compute_allocation_resource_subscribe.go | 33 +---
pkg/events/compute_allocation_subscribe.go | 33 +---
pkg/events/compute_cluster_user_subscribe.go | 33 +---
pkg/events/organization_subscribe.go | 33 +---
pkg/events/project_subscribe.go | 33 +---
pkg/events/types.go | 11 +-
pkg/events/user_identity_subscribe.go | 33 +---
pkg/events/user_subscribe.go | 33 +---
pkg/service/identity_resolver_integration_test.go | 3 +-
28 files changed, 350 insertions(+), 473 deletions(-)
diff --git a/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go b/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go
index 2d0ccdcea..296c93d8c 100644
--- a/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go
+++ b/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go
@@ -54,6 +54,9 @@ func init() {
const connectorName = "amie"
+// Type is the connector type in the config file.
+const Type = "amie-processor"
+
// LoadConnector skips silently when AMIE_BASE_URL / AMIE_SITE_CODE /
// AMIE_API_KEY are not all set. When enabled, it attaches /connectors/amie/*
// endpoints to mux.
diff --git a/connectors/Analytics/pkg/analytics/loader.go b/connectors/Analytics/pkg/analytics/loader.go
index 6cc7aa042..0c34609cf 100644
--- a/connectors/Analytics/pkg/analytics/loader.go
+++ b/connectors/Analytics/pkg/analytics/loader.go
@@ -35,6 +35,9 @@ import (
coreservice "github.com/apache/airavata-custos/pkg/service"
)
+// Type is the connector type in the config file.
+const Type = "analytics"
+
func LoadConnector(ctx context.Context, database *sqlx.DB, eventBus *events.Bus, coreService *coreservice.Service, wg *sync.WaitGroup, router *identity.Router, connectorConfig *config.ConnectorConfig) error {
svc := NewService(coreService, database)
NewHandlers(svc).RegisterRoutes(router)
diff --git a/connectors/COmanage/Identity-Provisioner/internal/subscribers/cluster_user.go b/connectors/COmanage/Identity-Provisioner/internal/subscribers/cluster_user.go
index e83e3d7fc..71a554192 100644
--- a/connectors/COmanage/Identity-Provisioner/internal/subscribers/cluster_user.go
+++ b/connectors/COmanage/Identity-Provisioner/internal/subscribers/cluster_user.go
@@ -53,20 +53,21 @@ func NewClusterUserSubscriber(c *client.Client, bus *events.Bus, core *service.S
}
}
-func (s *ClusterUserSubscriber) RegisterSubscribers() {
- s.bus.SubscribeComputeClusterUserCreated(s.handleClusterUserCreate)
- s.bus.SubscribeUserIdentityCreated(s.handleUserIdentityCreate)
+// RegisterSubscribers subscribes the handlers under the given subscriber name.
+func (s *ClusterUserSubscriber) RegisterSubscribers(subscriber string) {
+ s.bus.SubscribeComputeClusterUserCreated(subscriber, s.handleClusterUserCreate)
+ s.bus.SubscribeUserIdentityCreated(subscriber, s.handleUserIdentityCreate)
}
// handleUserIdentityCreate re-runs provisioning once the `User` has a `sub`.
// A user provisioned before their first sign-in has none in the registry, so
// the cluster cannot match the person to their ssh login.
-func (s *ClusterUserSubscriber) handleUserIdentityCreate(ctx context.Context, identity models.UserIdentity) {
+func (s *ClusterUserSubscriber) handleUserIdentityCreate(ctx context.Context, identity models.UserIdentity) error {
// Provisioning stores a COmanage identity of its own, which comes back with
// the `identity`. The event is fired for every `UserIdentity` source, and only the `oidc` one
// carries the sub that needs to be updated in the COmanage registry.
if identity.Source != "oidc" || identity.OIDCSub == "" {
- return
+ return nil
}
ctx = audit.WithSource(ctx, "comanage")
@@ -78,26 +79,28 @@ func (s *ClusterUserSubscriber) handleUserIdentityCreate(ctx context.Context, id
if errors.Is(err, service.ErrNotFound) {
// Expected for portal admins, system users, and temp users, which have no compute cluster account.
slog.Debug("comanage subscriber: no cluster account to link the sub to", "user_id", identity.UserID, "cluster_id", s.custosClusterID)
- return
+ return nil
}
if err != nil {
slog.Error("comanage subscriber: cluster user lookup failed", "user_id", identity.UserID, "err", err)
- return
+ return err
}
if err := s.ops.EnsurePOSIXAccount(ctx, cu); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
slog.Error("comanage subscriber: EnsurePOSIXAccount failed after identity link", "compute_cluster_user_id", cu.ID, "user_id", identity.UserID, "err", err)
+ return err
}
+ return nil
}
-func (s *ClusterUserSubscriber) handleClusterUserCreate(ctx context.Context, cu models.ComputeClusterUser) {
+func (s *ClusterUserSubscriber) handleClusterUserCreate(ctx context.Context, cu models.ComputeClusterUser) error {
ctx = audit.WithSource(ctx, "comanage")
ctx, span := tracing.Start(ctx, "comanage.cluster_user_create")
defer span.End()
if cu.ComputeClusterID != s.custosClusterID {
- return
+ return nil
}
span.SetAttributes(
attribute.String("comanage.cluster_user_id", cu.ID),
@@ -116,7 +119,7 @@ func (s *ClusterUserSubscriber) handleClusterUserCreate(ctx context.Context, cu
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
slog.Error("comanage subscriber: EnsurePOSIXAccount failed", "compute_cluster_user_id", cu.ID, "user_id", cu.UserID, "err", err)
- return
+ return err
}
// Even if this fails, the cluster account is usable without sudo, so a
// failed group join is logged and left for a retry.
@@ -124,6 +127,8 @@ func (s *ClusterUserSubscriber) handleClusterUserCreate(ctx context.Context, cu
if err := s.ops.EnsureClusterAdminMembership(ctx, &cu); err != nil {
span.RecordError(err)
slog.Error("comanage subscriber: EnsureClusterAdminMembership failed", "compute_cluster_user_id", cu.ID, "user_id", cu.UserID, "err", err)
+ return err
}
}
+ return nil
}
diff --git a/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go b/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go
index fabd12afc..f3746a706 100644
--- a/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go
+++ b/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go
@@ -43,6 +43,9 @@ func init() {
tracing.RegisterTerminalMarkers("comanage", "ComanageClusterAccountAttached")
}
+// Type is the connector type in the config file and the subscriber name on the bus.
+const Type = "comanage-identity-provisioner"
+
// LoadConnector wires the subscriber to the event bus. Reads YAML config first
// and falls back to environment variables. If neither yields a complete
// config, it logs and returns nil without registering.
@@ -56,7 +59,7 @@ func LoadConnector(_ context.Context, _ *sqlx.DB, eventBus *events.Bus, coreServ
}
}
httpClient := client.New(cfg)
- subscribers.NewClusterUserSubscriber(httpClient, eventBus, coreService, cfg.CustosClusterID).RegisterSubscribers()
+ subscribers.NewClusterUserSubscriber(httpClient, eventBus, coreService, cfg.CustosClusterID).RegisterSubscribers(Type)
slog.Info("comanage provisioner: subscriber registered",
"registry", cfg.RegistryURL, "co_id", cfg.COID, "cluster_id", cfg.CustosClusterID)
// Custos is the source of truth: CoPerson and UnixClusterAccount records
diff --git a/connectors/SLURM/Association-Mapper/internal/subscribers/account.go b/connectors/SLURM/Association-Mapper/internal/subscribers/account.go
index 6f20a1e47..cbb7c3b07 100644
--- a/connectors/SLURM/Association-Mapper/internal/subscribers/account.go
+++ b/connectors/SLURM/Association-Mapper/internal/subscribers/account.go
@@ -30,7 +30,7 @@ import (
"go.opentelemetry.io/otel/codes"
)
-func (a *AssociationSubscriber) SubscribeToComputeAllocationCreation(ctx context.Context, computeAllocation models.ComputeAllocation) {
+func (a *AssociationSubscriber) SubscribeToComputeAllocationCreation(ctx context.Context, computeAllocation models.ComputeAllocation) error {
ctx = audit.WithSource(ctx, "slurm")
ctx, span := tracing.Start(ctx, "slurm.compute_allocation_create")
defer span.End()
@@ -50,7 +50,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationCreation(ctx context
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationCreationFailed", "compute_allocation", computeAllocation.ID, "Failed to get compute cluster. Error: "+err.Error())
- return
+ return err
}
project, err := a.coreService.GetProject(ctx, computeAllocation.ProjectID)
@@ -59,7 +59,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationCreation(ctx context
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationCreationFailed", "compute_allocation", computeAllocation.ID, "Failed to get project. Error: "+err.Error())
- return
+ return err
}
pi, err := a.coreService.GetUser(ctx, project.ProjectPIID)
@@ -68,7 +68,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationCreation(ctx context
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationCreationFailed", "compute_allocation", computeAllocation.ID, "Failed to get project PI. Error: "+err.Error())
- return
+ return err
}
organization, err := a.coreService.GetOrganization(ctx, pi.OrganizationID)
@@ -77,7 +77,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationCreation(ctx context
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationCreationFailed", "compute_allocation", computeAllocation.ID, "Failed to get organization. Error: "+err.Error())
- return
+ return err
}
slurmAccount := client.Account{
@@ -92,17 +92,18 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationCreation(ctx context
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationCreationFailed", "compute_allocation", computeAllocation.ID, "Failed to create SLURM account. Error: "+err.Error())
- return
+ return err
}
a.recordAuditEvent(ctx, "ComputeAllocationCreationSucceeded", "compute_allocation", computeAllocation.ID, "Successfully created SLURM account for compute allocation.")
slog.Info("Successfully created SLURM account for compute allocation", "account", slurmAccount)
+ return nil
}
// SubscribeToComputeAllocationDeletion revokes every member's access when the
// allocation is deleted. The Slurm account is left in place so its job history
// stays attributable.
-func (a *AssociationSubscriber) SubscribeToComputeAllocationDeletion(ctx context.Context, computeAllocation models.ComputeAllocation) {
+func (a *AssociationSubscriber) SubscribeToComputeAllocationDeletion(ctx context.Context, computeAllocation models.ComputeAllocation) error {
ctx = audit.WithSource(ctx, "slurm")
ctx, span := tracing.Start(ctx, "slurm.compute_allocation_delete")
defer span.End()
@@ -116,15 +117,16 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationDeletion(ctx context
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationDeletionFailed", "compute_allocation", computeAllocation.ID, "Failed to remove associations. Error: "+err.Error())
- return
+ return err
}
a.recordAuditEvent(ctx, "ComputeAllocationDeprovisioned", "compute_allocation", computeAllocation.ID, "Removed cluster associations for the deleted allocation.")
+ return nil
}
// SubscribeToComputeAllocationUpdate keeps cluster access in sync with the
// allocation. An allocation that stops being active loses every association,
// and one that becomes active again has them restored.
-func (a *AssociationSubscriber) SubscribeToComputeAllocationUpdate(ctx context.Context, computeAllocation models.ComputeAllocation) {
+func (a *AssociationSubscriber) SubscribeToComputeAllocationUpdate(ctx context.Context, computeAllocation models.ComputeAllocation) error {
ctx = audit.WithSource(ctx, "slurm")
ctx, span := tracing.Start(ctx, "slurm.compute_allocation_update")
defer span.End()
@@ -139,10 +141,10 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationUpdate(ctx context.C
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationUpdateFailed", "compute_allocation", computeAllocation.ID, "Failed to restore associations. Error: "+err.Error())
- return
+ return err
}
a.recordAuditEvent(ctx, "ComputeAllocationProvisioned", "compute_allocation", computeAllocation.ID, "Restored cluster associations for the active allocation.")
- return
+ return nil
}
if err := a.removeAssociationsForAllocation(fetchCtx, computeAllocation); err != nil {
@@ -150,12 +152,13 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationUpdate(ctx context.C
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationUpdateFailed", "compute_allocation", computeAllocation.ID, "Failed to remove associations. Error: "+err.Error())
- return
+ return err
}
a.recordAuditEvent(ctx, "ComputeAllocationDeprovisioned", "compute_allocation", computeAllocation.ID, "Removed cluster associations for the deactivated allocation.")
+ return nil
}
-func (a *AssociationSubscriber) SubscribeToComputeAllocationResourceMappingCreation(ctx context.Context, mapping models.ComputeAllocationResourceMapping) {
+func (a *AssociationSubscriber) SubscribeToComputeAllocationResourceMappingCreation(ctx context.Context, mapping models.ComputeAllocationResourceMapping) error {
ctx = audit.WithSource(ctx, "slurm")
ctx, span := tracing.Start(ctx, "slurm.compute_allocation_resource_mapping_create")
defer span.End()
@@ -172,7 +175,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationResourceMappingCreat
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationResourceMappingCreationFailed", "compute_allocation_resource_mapping", mapping.ID, "Failed to get compute allocation. Error: "+err.Error())
- return
+ return err
}
cluster, err := a.coreService.GetComputeCluster(ctx, allocation.ComputeClusterID)
@@ -181,7 +184,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationResourceMappingCreat
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationResourceMappingCreationFailed", "compute_allocation_resource_mapping", mapping.ID, "Failed to get compute cluster. Error: "+err.Error())
- return
+ return err
}
span.SetAttributes(attribute.String("slurm.cluster_id", cluster.ID))
@@ -191,7 +194,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationResourceMappingCreat
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationResourceMappingCreationFailed", "compute_allocation_resource_mapping", mapping.ID, "Failed to get compute allocation resource. Error: "+err.Error())
- return
+ return err
}
grpTres := []client.TRES{}
@@ -202,7 +205,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationResourceMappingCreat
slog.Error("Resource type is empty for resource mapping creation", "mapping", mapping)
span.SetStatus(codes.Error, "resource type is empty")
a.recordAuditEvent(ctx, "ComputeAllocationResourceMappingCreationFailed", "compute_allocation_resource_mapping", mapping.ID, "Resource type is empty for resource mapping creation")
- return
+ return errEmptyResourceType
}
grpTres = append(grpTres, tresFor(resource.ResourceType, mapping.ResourceAmount))
@@ -215,7 +218,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationResourceMappingCreat
slog.Error("Resource type is empty for resource mapping creation", "mapping", mapping)
span.SetStatus(codes.Error, "resource type is empty")
a.recordAuditEvent(ctx, "ComputeAllocationResourceMappingCreationFailed", "compute_allocation_resource_mapping", mapping.ID, "Resource type is empty for resource mapping creation")
- return
+ return errEmptyResourceType
}
grpTresMins = append(grpTresMins, tresFor(resource.ResourceType, mapping.ResourceTime))
@@ -238,9 +241,10 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationResourceMappingCreat
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationResourceMappingCreationFailed", "compute_allocation_resource_mapping", mapping.ID, "Failed to upsert association. Error: "+err.Error())
- return
+ return err
} else {
slog.Info("Successfully upserted association for membership resource override creation", "association", association)
a.recordAuditEvent(ctx, "ComputeAllocationResourceMappingCreationSucceeded", "compute_allocation_resource_mapping", mapping.ID, "Successfully upserted association for compute allocation resource mapping creation")
}
+ return nil
}
diff --git a/connectors/SLURM/Association-Mapper/internal/subscribers/association_writer.go b/connectors/SLURM/Association-Mapper/internal/subscribers/association_writer.go
index af5676d02..09cca3279 100644
--- a/connectors/SLURM/Association-Mapper/internal/subscribers/association_writer.go
+++ b/connectors/SLURM/Association-Mapper/internal/subscribers/association_writer.go
@@ -43,6 +43,9 @@ func tresFor(resourceType string, count int64) client.TRES {
// and reject the user's jobs. The reconciler retries these.
var errNotProvisioned = errors.New("cluster account not provisioned yet")
+// errEmptyResourceType means the resource row has an empty resource_type.
+var errEmptyResourceType = errors.New("resource type is empty")
+
// assocKey identifies an association the way Slurm does. Cluster is not part
// of it because callers scope their lookups to one cluster already.
type assocKey struct {
diff --git a/connectors/SLURM/Association-Mapper/internal/subscribers/members.go b/connectors/SLURM/Association-Mapper/internal/subscribers/members.go
index e12be9744..1c9c68b4f 100644
--- a/connectors/SLURM/Association-Mapper/internal/subscribers/members.go
+++ b/connectors/SLURM/Association-Mapper/internal/subscribers/members.go
@@ -37,7 +37,7 @@ const (
reconcileSweepTimeout = 5 * time.Minute
)
-func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipCreation(ctx context.Context, membership models.ComputeAllocationMembership) {
+func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipCreation(ctx context.Context, membership models.ComputeAllocationMembership) error {
ctx = audit.WithSource(ctx, "slurm")
ctx, span := tracing.Start(ctx, "slurm.compute_allocation_membership_create")
defer span.End()
@@ -59,7 +59,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipCreation(c
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationMembershipCreationFailed", "compute_allocation_membership", membership.ID, "Failed to get compute allocation. Error: "+err.Error())
- return
+ return err
}
cluster, err := a.coreService.GetComputeCluster(fetchCtx, allocation.ComputeClusterID)
@@ -68,7 +68,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipCreation(c
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationMembershipCreationFailed", "compute_allocation_membership", membership.ID, "Failed to get compute cluster. Error: "+err.Error())
- return
+ return err
}
span.SetAttributes(attribute.String("slurm.cluster_id", cluster.ID))
@@ -80,18 +80,19 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipCreation(c
// flight. The reconciler writes it once the account lands.
slog.Info("Cluster account not provisioned yet, leaving the association to the reconciler",
"membership_id", membership.ID, "user_id", membership.UserID)
- return
+ return nil
}
slog.Error("Failed to upsert association for membership", "error", err)
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationMembershipCreationFailed", "compute_allocation_membership", membership.ID, "Failed to upsert association. Error: "+err.Error())
- return
+ return err
}
a.recordAuditEvent(ctx, "ComputeAllocationMembershipCreationSucceeded", "compute_allocation_membership", membership.ID, "Successfully upserted association.")
+ return nil
}
-func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipResourceOverrideCreation(ctx context.Context, override models.ComputeAllocationMembershipResourceOverride) {
+func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipResourceOverrideCreation(ctx context.Context, override models.ComputeAllocationMembershipResourceOverride) error {
ctx = audit.WithSource(ctx, "slurm")
ctx, span := tracing.Start(ctx, "slurm.compute_allocation_membership_resource_override_create")
defer span.End()
@@ -107,7 +108,7 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipResourceOv
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationMembershipResourceOverrideCreationFailed", "compute_allocation_membership_resource_override", override.ID, "Failed to get compute allocation membership. Error: "+err.Error())
- return
+ return err
}
span.SetAttributes(
attribute.String("slurm.allocation_id", membership.ComputeAllocationID),
@@ -121,21 +122,22 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipResourceOv
if errors.Is(err, errNotProvisioned) {
slog.Info("Cluster account not provisioned yet, leaving the association to the reconciler",
"membership_id", membership.ID, "override_id", override.ID)
- return
+ return nil
}
slog.Error("Failed to upsert association for membership resource override", "error", err)
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationMembershipResourceOverrideCreationFailed", "compute_allocation_membership_resource_override", override.ID, "Failed to upsert association. Error: "+err.Error())
- return
+ return err
}
a.recordAuditEvent(ctx, "ComputeAllocationMembershipResourceOverrideCreationSucceeded", "compute_allocation_membership_resource_override", override.ID, "Successfully upserted association.")
+ return nil
}
// SubscribeToComputeAllocationMembershipUpdate reacts to a membership-changing
// state. Anything other than ACTIVE means the member should no longer be able
// to submit, so the associations go; going back to ACTIVE re-grants them.
-func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipUpdate(ctx context.Context, membership models.ComputeAllocationMembership) {
+func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipUpdate(ctx context.Context, membership models.ComputeAllocationMembership) error {
ctx = audit.WithSource(ctx, "slurm")
ctx, span := tracing.Start(ctx, "slurm.compute_allocation_membership_update")
defer span.End()
@@ -152,14 +154,15 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipUpdate(ctx
if errors.Is(err, errNotProvisioned) {
slog.Info("Cluster account not provisioned yet, leaving the association to the reconciler",
"membership_id", membership.ID)
- return
+ return nil
}
slog.Error("Failed to upsert association for reactivated membership", "error", err)
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationMembershipUpdateFailed", "compute_allocation_membership", membership.ID, "Failed to upsert association. Error: "+err.Error())
+ return err
}
- return
+ return nil
}
if err := a.removeAssociationsForMembership(fetchCtx, membership); err != nil {
@@ -167,14 +170,15 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipUpdate(ctx
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationMembershipUpdateFailed", "compute_allocation_membership", membership.ID, "Failed to remove associations. Error: "+err.Error())
- return
+ return err
}
a.recordAuditEvent(ctx, "ComputeAllocationMembershipDeprovisioned", "compute_allocation_membership", membership.ID, "Removed cluster associations for the deactivated membership.")
+ return nil
}
// SubscribeToComputeAllocationMembershipDeletion removes the member's access
// when the membership row itself goes away.
-func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipDeletion(ctx context.Context, membership models.ComputeAllocationMembership) {
+func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipDeletion(ctx context.Context, membership models.ComputeAllocationMembership) error {
ctx = audit.WithSource(ctx, "slurm")
ctx, span := tracing.Start(ctx, "slurm.compute_allocation_membership_delete")
defer span.End()
@@ -191,7 +195,8 @@ func (a *AssociationSubscriber) SubscribeToComputeAllocationMembershipDeletion(c
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordAuditEvent(ctx, "ComputeAllocationMembershipDeletionFailed", "compute_allocation_membership", membership.ID, "Failed to remove associations. Error: "+err.Error())
- return
+ return err
}
a.recordAuditEvent(ctx, "ComputeAllocationMembershipDeprovisioned", "compute_allocation_membership", membership.ID, "Removed cluster associations for the deleted membership.")
+ return nil
}
diff --git a/connectors/SLURM/Association-Mapper/internal/subscribers/members_test.go b/connectors/SLURM/Association-Mapper/internal/subscribers/members_test.go
index f25e98cd5..d3be7f56d 100644
--- a/connectors/SLURM/Association-Mapper/internal/subscribers/members_test.go
+++ b/connectors/SLURM/Association-Mapper/internal/subscribers/members_test.go
@@ -19,6 +19,7 @@ package subscribers
import (
"context"
+ "errors"
"sync"
"testing"
"time"
@@ -156,8 +157,11 @@ func testMembership() models.ComputeAllocationMembership {
func TestMembershipCreationWritesAssociationWhenProvisioned(t *testing.T) {
core := coreMock(mockOpts{provisionedAt: ago(time.Minute)})
slurm := &fakeSlurmClient{}
- NewAssociationSubscriber(slurm, nil, core, 0, 0).
+ err := NewAssociationSubscriber(slurm, nil, core, 0, 0).
SubscribeToComputeAllocationMembershipCreation(context.Background(), testMembership())
+ if err != nil {
+ t.Fatalf("expected no error, got %v", err)
+ }
got := slurm.all()
if len(got) != 1 {
@@ -168,13 +172,33 @@ func TestMembershipCreationWritesAssociationWhenProvisioned(t *testing.T) {
}
}
+// Make sure a failed core lookup is returned as the handler's error and not
+// only logged, so the bus can tell the delivery failed.
+func TestMembershipCreationReturnsCoreError(t *testing.T) {
+ core := coreMock(mockOpts{provisionedAt: ago(time.Minute)})
+ lookupErr := errors.New("core unavailable")
+ core.GetComputeAllocationFunc = func(context.Context, string) (*models.ComputeAllocation, error) {
+ return nil, lookupErr
+ }
+
+ err := NewAssociationSubscriber(&fakeSlurmClient{}, nil, core, 0, 0).
+ SubscribeToComputeAllocationMembershipCreation(context.Background(), testMembership())
+ if !errors.Is(err, lookupErr) {
+ t.Fatalf("expected the core lookup error, got %v", err)
+ }
+}
+
// An unprovisioned account must NOT get an association: slurmctld would cache
-// the failed uid lookup and reject the user's jobs. The reconciler picks it up.
+// the failed uid lookup and reject the user's jobs. The reconciler picks it up,
+// so the handler returns no error.
func TestMembershipCreationDefersWhenNotProvisioned(t *testing.T) {
core := coreMock(mockOpts{provisionedAt: nil})
slurm := &fakeSlurmClient{}
- NewAssociationSubscriber(slurm, nil, core, 0, 0).
+ err := NewAssociationSubscriber(slurm, nil, core, 0, 0).
SubscribeToComputeAllocationMembershipCreation(context.Background(), testMembership())
+ if err != nil {
+ t.Fatalf("expected no error for an unprovisioned account, got %v", err)
+ }
if n := len(slurm.all()); n != 0 {
t.Fatalf("expected no association for an unprovisioned account, got %d", n)
diff --git a/connectors/SLURM/Association-Mapper/internal/subscribers/subscriber.go b/connectors/SLURM/Association-Mapper/internal/subscribers/subscriber.go
index 4606c1b79..63af184f9 100644
--- a/connectors/SLURM/Association-Mapper/internal/subscribers/subscriber.go
+++ b/connectors/SLURM/Association-Mapper/internal/subscribers/subscriber.go
@@ -53,15 +53,16 @@ func NewAssociationSubscriber(slurmClient SlurmClient, eventBus *events.Bus, cor
}
}
-func (a *AssociationSubscriber) RegisterSubscribers() {
- a.eventBus.SubscribeComputeAllocationCreated(a.SubscribeToComputeAllocationCreation)
- a.eventBus.SubscribeComputeAllocationDeleted(a.SubscribeToComputeAllocationDeletion)
- a.eventBus.SubscribeComputeAllocationUpdated(a.SubscribeToComputeAllocationUpdate)
- a.eventBus.SubscribeComputeAllocationMembershipCreated(a.SubscribeToComputeAllocationMembershipCreation)
- a.eventBus.SubscribeComputeAllocationMembershipUpdated(a.SubscribeToComputeAllocationMembershipUpdate)
- a.eventBus.SubscribeComputeAllocationMembershipDeleted(a.SubscribeToComputeAllocationMembershipDeletion)
- a.eventBus.SubscribeComputeAllocationMembershipResourceOverrideCreated(a.SubscribeToComputeAllocationMembershipResourceOverrideCreation)
- a.eventBus.SubscribeComputeAllocationResourceMappingCreated(a.SubscribeToComputeAllocationResourceMappingCreation)
+// RegisterSubscribers subscribes the handlers under the given subscriber name.
+func (a *AssociationSubscriber) RegisterSubscribers(subscriber string) {
+ a.eventBus.SubscribeComputeAllocationCreated(subscriber, a.SubscribeToComputeAllocationCreation)
+ a.eventBus.SubscribeComputeAllocationDeleted(subscriber, a.SubscribeToComputeAllocationDeletion)
+ a.eventBus.SubscribeComputeAllocationUpdated(subscriber, a.SubscribeToComputeAllocationUpdate)
+ a.eventBus.SubscribeComputeAllocationMembershipCreated(subscriber, a.SubscribeToComputeAllocationMembershipCreation)
+ a.eventBus.SubscribeComputeAllocationMembershipUpdated(subscriber, a.SubscribeToComputeAllocationMembershipUpdate)
+ a.eventBus.SubscribeComputeAllocationMembershipDeleted(subscriber, a.SubscribeToComputeAllocationMembershipDeletion)
+ a.eventBus.SubscribeComputeAllocationMembershipResourceOverrideCreated(subscriber, a.SubscribeToComputeAllocationMembershipResourceOverrideCreation)
+ a.eventBus.SubscribeComputeAllocationResourceMappingCreated(subscriber, a.SubscribeToComputeAllocationResourceMappingCreation)
}
func (a *AssociationSubscriber) recordAuditEvent(ctx context.Context, eventType, entityType, entityId, message string) {
diff --git a/connectors/SLURM/Association-Mapper/pkg/smapper/loader.go b/connectors/SLURM/Association-Mapper/pkg/smapper/loader.go
index c3bda7a25..128474128 100644
--- a/connectors/SLURM/Association-Mapper/pkg/smapper/loader.go
+++ b/connectors/SLURM/Association-Mapper/pkg/smapper/loader.go
@@ -34,6 +34,9 @@ import (
"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 = "slurm-association-mapper"
+
func LoadConnector(ctx context.Context, _ *sqlx.DB, eventBus *events.Bus, coreService *service.Service, wg *sync.WaitGroup, _ *identity.Router, connectorConfig *config.ConnectorConfig) error {
// Read url, username, and password from config or environment variables
@@ -94,7 +97,7 @@ func LoadConnector(ctx context.Context, _ *sqlx.DB, eventBus *events.Bus, coreSe
slurmClient := client.New(apiUrl, user, token, apiVersion)
subscriber := subscribers.NewAssociationSubscriber(slurmClient, eventBus, coreService, reconcileInterval, provisionGrace)
- subscriber.RegisterSubscribers()
+ subscriber.RegisterSubscribers(Type)
// The sweep, not the events above, is what guarantees associations exist.
wg.Add(1)
diff --git a/connectors/SLURM/Usage-Monitor/pkg/monitor/loader.go b/connectors/SLURM/Usage-Monitor/pkg/monitor/loader.go
index 1632f5fdf..1744f4281 100644
--- a/connectors/SLURM/Usage-Monitor/pkg/monitor/loader.go
+++ b/connectors/SLURM/Usage-Monitor/pkg/monitor/loader.go
@@ -33,6 +33,9 @@ import (
"github.com/jmoiron/sqlx"
)
+// Type is the connector type in the config file.
+const Type = "slurm-usage-monitor"
+
func LoadConnector(ctx context.Context, _ *sqlx.DB, eventBus *events.Bus, coreService *service.Service, wg *sync.WaitGroup, _ *identity.Router, connectorConfig *config.ConnectorConfig) error {
// Read url, username, and password from config or environment variables
diff --git a/connectors/TempAccount/pkg/tempaccount/loader.go b/connectors/TempAccount/pkg/tempaccount/loader.go
index ae42a0cf9..c3d163a87 100644
--- a/connectors/TempAccount/pkg/tempaccount/loader.go
+++ b/connectors/TempAccount/pkg/tempaccount/loader.go
@@ -29,6 +29,9 @@ import (
"github.com/jmoiron/sqlx"
)
+// Type is the connector type in the config file.
+const Type = "temp-account"
+
func LoadConnector(ctx context.Context, _ *sqlx.DB, eventBus *events.Bus, coreService *service.Service, wg *sync.WaitGroup, router *identity.Router, connectorConfig *config.ConnectorConfig) error {
handlers := internal.NewHandlers(coreService)
handlers.RegisterRoutes(router)
diff --git a/internal/connectors/loader.go b/internal/connectors/loader.go
index 9c41f0c62..5e34b88f7 100644
--- a/internal/connectors/loader.go
+++ b/internal/connectors/loader.go
@@ -39,12 +39,12 @@ func LoadConnectorsFromConfig(ctx context.Context, cfg *config.Config, database
slog.Info("loading connectors from config")
connectorLoaders := map[string]func(context.Context, *sqlx.DB, *events.Bus, *service.Service, *sync.WaitGroup, *identity.Router, *config.ConnectorConfig) error{
- "slurm-association-mapper": smapper.LoadConnector,
- "amie-processor": amie.LoadConnector,
- "comanage-identity-provisioner": comanage.LoadConnector,
- "slurm-usage-monitor": monitor.LoadConnector,
- "temp-account": tempaccount.LoadConnector,
- "analytics": analytics.LoadConnector,
+ smapper.Type: smapper.LoadConnector,
+ amie.Type: amie.LoadConnector,
+ comanage.Type: comanage.LoadConnector,
+ monitor.Type: monitor.LoadConnector,
+ tempaccount.Type: tempaccount.LoadConnector,
+ analytics.Type: analytics.LoadConnector,
}
for connectorName, connectorCfg := range cfg.Connectors {
diff --git a/pkg/events/bus.go b/pkg/events/bus.go
index 87a26cee9..7129f0723 100644
--- a/pkg/events/bus.go
+++ b/pkg/events/bus.go
@@ -29,17 +29,34 @@ import (
func New() *Bus {
return &Bus{
- subs: make(map[string][]EventSubscriberFunc),
+ subs: make(map[string][]subscription),
}
}
-// Subscribe registers a handler for a given topic.
+// Subscribe registers a handler for a given topic under the subscriber's name.
// The handler is called asynchronously (in a new goroutine) each time
// an event is published on that topic.
-func (b *Bus) Subscribe(topic EventType, handler EventSubscriberFunc) {
+func (b *Bus) Subscribe(subscriber string, topic EventType, handler EventSubscriberFunc) {
b.mu.Lock()
defer b.mu.Unlock()
- b.subs[string(topic)] = append(b.subs[string(topic)], handler)
+ b.subs[string(topic)] = append(b.subs[string(topic)], subscription{subscriber: subscriber, handler: handler})
+}
+
+// subscribeTyped registers a handler that takes the payload as T. The payload may be published as T or *T.
+func subscribeTyped[T any](b *Bus, subscriber string, topic EventType, handler func(context.Context, T) error) {
+ b.Subscribe(subscriber, topic, func(ctx context.Context, event Event, value interface{}) error {
+ switch v := value.(type) {
+ case T:
+ return handler(ctx, v)
+ case *T:
+ if v != nil {
+ return handler(ctx, *v)
+ }
+ default:
+ slog.Warn("event payload has unexpected type", "type", event.Type, "got", value)
+ }
+ return nil
+ })
}
// Publish sends an event to all subscribers of the given topic.
@@ -49,18 +66,18 @@ func (b *Bus) Publish(ctx context.Context, topic EventType, payload any) {
defer span.End()
b.mu.RLock()
- handlers := make([]EventSubscriberFunc, len(b.subs[string(topic)]))
- copy(handlers, b.subs[string(topic)])
+ subs := make([]subscription, len(b.subs[string(topic)]))
+ copy(subs, b.subs[string(topic)])
b.mu.RUnlock()
event := Event{Type: topic, Payload: payload}
detached := context.WithoutCancel(ctx)
- for _, h := range handlers {
- go safeDispatch(detached, h, event, payload)
+ for _, s := range subs {
+ go safeDispatch(detached, s, event, payload)
}
}
-func safeDispatch(ctx context.Context, h EventSubscriberFunc, event Event, payload any) {
+func safeDispatch(ctx context.Context, s subscription, event Event, payload any) {
ctx, span := tracing.Start(ctx, "bus.subscribe:"+string(event.Type))
defer span.End()
@@ -71,32 +88,45 @@ func safeDispatch(ctx context.Context, h EventSubscriberFunc, event Event, paylo
span.SetStatus(codes.Error, "subscriber panic")
slog.Error("event subscriber panicked",
"topic", event.Type,
+ "subscriber", s.subscriber,
"panic", r,
"stack", string(debug.Stack()),
)
}
}()
- h(ctx, event, payload)
+ if err := s.handler(ctx, event, payload); err != nil {
+ span.RecordError(err)
+ span.SetStatus(codes.Error, err.Error())
+ slog.ErrorContext(ctx, "event subscriber failed",
+ "topic", event.Type,
+ "subscriber", s.subscriber,
+ "error", err,
+ )
+ }
}
// PublishSync is like Publish but calls handlers in the caller's goroutine.
// Useful when you need to guarantee ordering or want backpressure.
-func (b *Bus) PublishSync(ctx context.Context, topic EventType, payload any) {
+// It stops at the first handler error and returns it.
+func (b *Bus) PublishSync(ctx context.Context, topic EventType, payload any) error {
ctx, span := tracing.Start(ctx, "bus.publish:"+string(topic))
defer span.End()
b.mu.RLock()
- handlers := make([]EventSubscriberFunc, len(b.subs[string(topic)]))
- copy(handlers, b.subs[string(topic)])
+ subs := make([]subscription, len(b.subs[string(topic)]))
+ copy(subs, b.subs[string(topic)])
b.mu.RUnlock()
event := Event{Type: topic, Payload: payload}
- for _, h := range handlers {
- dispatchSync(ctx, h, event, payload)
+ for _, s := range subs {
+ if err := dispatchSync(ctx, s, event, payload); err != nil {
+ return err
+ }
}
+ return nil
}
-func dispatchSync(ctx context.Context, h EventSubscriberFunc, event Event, payload any) {
+func dispatchSync(ctx context.Context, s subscription, event Event, payload any) error {
ctx, span := tracing.Start(ctx, "bus.subscribe:"+string(event.Type))
defer span.End()
@@ -107,11 +137,17 @@ func dispatchSync(ctx context.Context, h EventSubscriberFunc, event Event, paylo
span.SetStatus(codes.Error, "subscriber panic")
slog.Error("event subscriber panicked",
"topic", event.Type,
+ "subscriber", s.subscriber,
"panic", r,
"stack", string(debug.Stack()),
)
panic(r)
}
}()
- h(ctx, event, payload)
+ err := s.handler(ctx, event, payload)
+ if err != nil {
+ span.RecordError(err)
+ span.SetStatus(codes.Error, err.Error())
+ }
+ return err
}
diff --git a/pkg/events/bus_test.go b/pkg/events/bus_test.go
index 46af671a6..54b505e57 100644
--- a/pkg/events/bus_test.go
+++ b/pkg/events/bus_test.go
@@ -19,6 +19,7 @@ package events
import (
"context"
+ "errors"
"sync"
"sync/atomic"
"testing"
@@ -29,62 +30,18 @@ import (
sdktrace "go.opentelemetry.io/otel/sdk/trace"
)
-type ctxKey string
-
-const testCtxKey ctxKey = "trace-id"
-
const topicTest EventType = "test::topic"
-func TestPublishSyncPropagatesContext(t *testing.T) {
- bus := New()
-
- got := make(chan string, 1)
- bus.Subscribe(topicTest, func(ctx context.Context, _ Event, _ interface{}) {
- v, _ := ctx.Value(testCtxKey).(string)
- got <- v
- })
-
- ctx := context.WithValue(context.Background(), testCtxKey, "abc123")
- bus.PublishSync(ctx, topicTest, "payload")
-
- select {
- case v := <-got:
- if v != "abc123" {
- t.Fatalf("PublishSync did not propagate ctx value, got %q", v)
- }
- case <-time.After(time.Second):
- t.Fatalf("subscriber never ran")
- }
-}
-
-func TestPublishAsyncPropagatesContext(t *testing.T) {
- bus := New()
-
- got := make(chan string, 1)
- bus.Subscribe(topicTest, func(ctx context.Context, _ Event, _ interface{}) {
- v, _ := ctx.Value(testCtxKey).(string)
- got <- v
- })
-
- ctx := context.WithValue(context.Background(), testCtxKey, "def456")
- bus.Publish(ctx, topicTest, "payload")
+const subscriberTest = "test-subscriber"
- select {
- case v := <-got:
- if v != "def456" {
- t.Fatalf("Publish did not propagate ctx value, got %q", v)
- }
- case <-time.After(time.Second):
- t.Fatalf("subscriber never ran")
- }
-}
-
-func TestPublishAsyncDetachesCancellation(t *testing.T) {
+// Make sure a handler is not canceled when the publisher's request context is
+// canceled, since a handler can keep running after publishing the event.
+func TestPublishDetachesCancellation(t *testing.T) {
bus := New()
started := make(chan struct{})
done := make(chan error, 1)
- bus.Subscribe(topicTest, func(ctx context.Context, _ Event, _ interface{}) {
+ bus.Subscribe(subscriberTest, topicTest, func(ctx context.Context, _ Event, _ interface{}) error {
close(started)
select {
case <-ctx.Done():
@@ -92,6 +49,7 @@ func TestPublishAsyncDetachesCancellation(t *testing.T) {
case <-time.After(200 * time.Millisecond):
done <- nil
}
+ return nil
})
ctx, cancel := context.WithCancel(context.Background())
@@ -110,38 +68,22 @@ func TestPublishAsyncDetachesCancellation(t *testing.T) {
}
}
-func TestPublishSyncPanicPropagatesToCaller(t *testing.T) {
- bus := New()
-
- bus.Subscribe(topicTest, func(context.Context, Event, interface{}) {
- panic("boom")
- })
-
- var recovered any
- func() {
- defer func() { recovered = recover() }()
- bus.PublishSync(context.Background(), topicTest, nil)
- }()
-
- if recovered == nil {
- t.Fatalf("expected sync publish to surface subscriber panic to caller")
- }
-}
-
-func TestPublishAsyncSubscriberPanicDoesNotKillOthers(t *testing.T) {
+// Make sure one subscriber panicking does not stop another subscriber of the same event.
+func TestPublishSubscriberPanicDoesNotKillOthers(t *testing.T) {
bus := New()
var ran atomic.Int32
var wg sync.WaitGroup
wg.Add(2)
- bus.Subscribe(topicTest, func(context.Context, Event, interface{}) {
+ bus.Subscribe(subscriberTest, topicTest, func(context.Context, Event, interface{}) error {
defer wg.Done()
ran.Add(1)
panic("boom")
})
- bus.Subscribe(topicTest, func(context.Context, Event, interface{}) {
+ bus.Subscribe(subscriberTest, topicTest, func(context.Context, Event, interface{}) error {
defer wg.Done()
ran.Add(1)
+ return nil
})
bus.Publish(context.Background(), topicTest, nil)
@@ -162,32 +104,9 @@ func TestPublishAsyncSubscriberPanicDoesNotKillOthers(t *testing.T) {
}
}
-type recordingProcessor struct {
- mu sync.Mutex
- spans []sdktrace.ReadOnlySpan
-}
-
-func (p *recordingProcessor) OnStart(context.Context, sdktrace.ReadWriteSpan) {}
-func (p *recordingProcessor) OnEnd(s sdktrace.ReadOnlySpan) {
- p.mu.Lock()
- defer p.mu.Unlock()
- p.spans = append(p.spans, s)
-}
-func (p *recordingProcessor) Shutdown(context.Context) error { return nil }
-func (p *recordingProcessor) ForceFlush(context.Context) error { return nil }
-
-func (p *recordingProcessor) findByName(name string) sdktrace.ReadOnlySpan {
- p.mu.Lock()
- defer p.mu.Unlock()
- for _, s := range p.spans {
- if s.Name() == name {
- return s
- }
- }
- return nil
-}
-
-func TestSafeDispatchSetsSpanErrorOnPanic(t *testing.T) {
+// Make sure a handler error marks the subscribe span as failed, so a failed
+// delivery does not show as healthy in the trace view.
+func TestPublishMarksSpanErrorWhenHandlerFails(t *testing.T) {
rec := &recordingProcessor{}
tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec))
prev := otel.GetTracerProvider()
@@ -195,8 +114,8 @@ func TestSafeDispatchSetsSpanErrorOnPanic(t *testing.T) {
t.Cleanup(func() { otel.SetTracerProvider(prev) })
bus := New()
- bus.Subscribe(topicTest, func(context.Context, Event, interface{}) {
- panic("boom")
+ bus.Subscribe(subscriberTest, topicTest, func(context.Context, Event, interface{}) error {
+ return errors.New("boom")
})
bus.Publish(context.Background(), topicTest, nil)
@@ -213,13 +132,76 @@ func TestSafeDispatchSetsSpanErrorOnPanic(t *testing.T) {
}
if sub == nil {
- var names []string
- for _, s := range rec.spans {
- names = append(names, s.Name())
- }
- t.Fatalf("did not see bus.subscribe span; got names=%v", names)
+ t.Fatalf("did not see bus.subscribe span")
}
if got := sub.Status().Code; got != codes.Error {
t.Fatalf("expected bus.subscribe span status=Error, got %v", got)
}
}
+
+// Make sure the `sync publish` stops at the first failing handler and returns its error.
+func TestPublishSyncReturnsHandlerError(t *testing.T) {
+ bus := New()
+
+ want := errors.New("boom")
+ var secondRan bool
+ bus.Subscribe(subscriberTest, topicTest, func(context.Context, Event, interface{}) error {
+ return want
+ })
+ bus.Subscribe(subscriberTest, topicTest, func(context.Context, Event, interface{}) error {
+ secondRan = true
+ return nil
+ })
+
+ if err := bus.PublishSync(context.Background(), topicTest, nil); !errors.Is(err, want) {
+ t.Fatalf("expected handler error, got %v", err)
+ }
+ if secondRan {
+ t.Fatalf("expected sync publish to stop at the first error")
+ }
+}
+
+// Make sure a handler panic reaches the caller of the `sync publish`, unlike
+// Publish method, which runs handlers in the background and can only log it.
+func TestPublishSyncPanicPropagatesToCaller(t *testing.T) {
+ bus := New()
+
+ bus.Subscribe(subscriberTest, topicTest, func(context.Context, Event, interface{}) error {
+ panic("boom")
+ })
+
+ var recovered any
+ func() {
+ defer func() { recovered = recover() }()
+ _ = bus.PublishSync(context.Background(), topicTest, nil)
+ }()
+
+ if recovered == nil {
+ t.Fatalf("expected sync publish to surface subscriber panic to caller")
+ }
+}
+
+type recordingProcessor struct {
+ mu sync.Mutex
+ spans []sdktrace.ReadOnlySpan
+}
+
+func (p *recordingProcessor) OnStart(context.Context, sdktrace.ReadWriteSpan) {}
+func (p *recordingProcessor) OnEnd(s sdktrace.ReadOnlySpan) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ p.spans = append(p.spans, s)
+}
+func (p *recordingProcessor) Shutdown(context.Context) error { return nil }
+func (p *recordingProcessor) ForceFlush(context.Context) error { return nil }
+
+func (p *recordingProcessor) findByName(name string) sdktrace.ReadOnlySpan {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ for _, s := range p.spans {
+ if s.Name() == name {
+ return s
+ }
+ }
+ return nil
+}
diff --git a/pkg/events/compute_allocation_diff_subscribe.go b/pkg/events/compute_allocation_diff_subscribe.go
index f4fd09b31..4b83637c4 100644
--- a/pkg/events/compute_allocation_diff_subscribe.go
+++ b/pkg/events/compute_allocation_diff_subscribe.go
@@ -19,48 +19,29 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// ComputeAllocationDiffHandler handles compute allocation diff lifecycle events with a typed payload.
-type ComputeAllocationDiffHandler func(ctx context.Context, diff models.ComputeAllocationDiff)
+type ComputeAllocationDiffHandler func(ctx context.Context, diff models.ComputeAllocationDiff) error
// SubscribeComputeAllocationDiffCreated registers a typed handler invoked whenever a
// compute_allocation_diff::create event is published. Events with payloads that are
// not a models.ComputeAllocationDiff (or *models.ComputeAllocationDiff) are dropped
// with a warning log.
-func (b *Bus) SubscribeComputeAllocationDiffCreated(handler ComputeAllocationDiffHandler) {
- b.subscribeComputeAllocationDiff(ComputeAllocationDiffCreateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationDiffCreated(subscriber string, handler ComputeAllocationDiffHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationDiffCreateEvent, handler)
}
// SubscribeComputeAllocationDiffUpdated registers a typed handler invoked whenever a
// compute_allocation_diff::update event is published.
-func (b *Bus) SubscribeComputeAllocationDiffUpdated(handler ComputeAllocationDiffHandler) {
- b.subscribeComputeAllocationDiff(ComputeAllocationDiffUpdateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationDiffUpdated(subscriber string, handler ComputeAllocationDiffHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationDiffUpdateEvent, handler)
}
// SubscribeComputeAllocationDiffDeleted registers a typed handler invoked whenever a
// compute_allocation_diff::delete event is published.
-func (b *Bus) SubscribeComputeAllocationDiffDeleted(handler ComputeAllocationDiffHandler) {
- b.subscribeComputeAllocationDiff(ComputeAllocationDiffDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeComputeAllocationDiff(topic EventType, handler ComputeAllocationDiffHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch d := value.(type) {
- case models.ComputeAllocationDiff:
- handler(ctx, d)
- case *models.ComputeAllocationDiff:
- if d != nil {
- handler(ctx, *d)
- }
- default:
- slog.Warn("compute allocation diff event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeComputeAllocationDiffDeleted(subscriber string, handler ComputeAllocationDiffHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationDiffDeleteEvent, handler)
}
diff --git a/pkg/events/compute_allocation_membership_resource_override_subscribe.go b/pkg/events/compute_allocation_membership_resource_override_subscribe.go
index 6a381a524..ae34323c1 100644
--- a/pkg/events/compute_allocation_membership_resource_override_subscribe.go
+++ b/pkg/events/compute_allocation_membership_resource_override_subscribe.go
@@ -19,50 +19,31 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// ComputeAllocationMembershipResourceOverrideHandler handles lifecycle
// events for membership resource overrides with a typed payload.
-type ComputeAllocationMembershipResourceOverrideHandler func(ctx context.Context, o models.ComputeAllocationMembershipResourceOverride)
+type ComputeAllocationMembershipResourceOverrideHandler func(ctx context.Context, o models.ComputeAllocationMembershipResourceOverride) error
// SubscribeComputeAllocationMembershipResourceOverrideCreated registers a
// typed handler invoked whenever a
// compute_allocation_membership_resource_override::create event is published.
-func (b *Bus) SubscribeComputeAllocationMembershipResourceOverrideCreated(handler ComputeAllocationMembershipResourceOverrideHandler) {
- b.subscribeMembershipResourceOverride(ComputeAllocationMembershipResourceOverrideCreateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationMembershipResourceOverrideCreated(subscriber string, handler ComputeAllocationMembershipResourceOverrideHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationMembershipResourceOverrideCreateEvent, handler)
}
// SubscribeComputeAllocationMembershipResourceOverrideUpdated registers a
// typed handler invoked whenever a
// compute_allocation_membership_resource_override::update event is published.
-func (b *Bus) SubscribeComputeAllocationMembershipResourceOverrideUpdated(handler ComputeAllocationMembershipResourceOverrideHandler) {
- b.subscribeMembershipResourceOverride(ComputeAllocationMembershipResourceOverrideUpdateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationMembershipResourceOverrideUpdated(subscriber string, handler ComputeAllocationMembershipResourceOverrideHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationMembershipResourceOverrideUpdateEvent, handler)
}
// SubscribeComputeAllocationMembershipResourceOverrideDeleted registers a
// typed handler invoked whenever a
// compute_allocation_membership_resource_override::delete event is published.
-func (b *Bus) SubscribeComputeAllocationMembershipResourceOverrideDeleted(handler ComputeAllocationMembershipResourceOverrideHandler) {
- b.subscribeMembershipResourceOverride(ComputeAllocationMembershipResourceOverrideDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeMembershipResourceOverride(topic EventType, handler ComputeAllocationMembershipResourceOverrideHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch o := value.(type) {
- case models.ComputeAllocationMembershipResourceOverride:
- handler(ctx, o)
- case *models.ComputeAllocationMembershipResourceOverride:
- if o != nil {
- handler(ctx, *o)
- }
- default:
- slog.Warn("compute allocation membership resource override event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeComputeAllocationMembershipResourceOverrideDeleted(subscriber string, handler ComputeAllocationMembershipResourceOverrideHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationMembershipResourceOverrideDeleteEvent, handler)
}
diff --git a/pkg/events/compute_allocation_membership_subscribe.go b/pkg/events/compute_allocation_membership_subscribe.go
index 169753013..632e8ec65 100644
--- a/pkg/events/compute_allocation_membership_subscribe.go
+++ b/pkg/events/compute_allocation_membership_subscribe.go
@@ -19,48 +19,29 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// ComputeAllocationMembershipHandler handles compute allocation membership lifecycle events with a typed payload.
-type ComputeAllocationMembershipHandler func(ctx context.Context, membership models.ComputeAllocationMembership)
+type ComputeAllocationMembershipHandler func(ctx context.Context, membership models.ComputeAllocationMembership) error
// SubscribeComputeAllocationMembershipCreated registers a typed handler invoked whenever a
// compute_allocation_membership::create event is published. Events with payloads that are
// not a models.ComputeAllocationMembership (or *models.ComputeAllocationMembership) are
// dropped with a warning log.
-func (b *Bus) SubscribeComputeAllocationMembershipCreated(handler ComputeAllocationMembershipHandler) {
- b.subscribeComputeAllocationMembership(ComputeAllocationMembershipCreateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationMembershipCreated(subscriber string, handler ComputeAllocationMembershipHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationMembershipCreateEvent, handler)
}
// SubscribeComputeAllocationMembershipUpdated registers a typed handler invoked whenever a
// compute_allocation_membership::update event is published.
-func (b *Bus) SubscribeComputeAllocationMembershipUpdated(handler ComputeAllocationMembershipHandler) {
- b.subscribeComputeAllocationMembership(ComputeAllocationMembershipUpdateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationMembershipUpdated(subscriber string, handler ComputeAllocationMembershipHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationMembershipUpdateEvent, handler)
}
// SubscribeComputeAllocationMembershipDeleted registers a typed handler invoked whenever a
// compute_allocation_membership::delete event is published.
-func (b *Bus) SubscribeComputeAllocationMembershipDeleted(handler ComputeAllocationMembershipHandler) {
- b.subscribeComputeAllocationMembership(ComputeAllocationMembershipDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeComputeAllocationMembership(topic EventType, handler ComputeAllocationMembershipHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch m := value.(type) {
- case models.ComputeAllocationMembership:
- handler(ctx, m)
- case *models.ComputeAllocationMembership:
- if m != nil {
- handler(ctx, *m)
- }
- default:
- slog.Warn("compute allocation membership event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeComputeAllocationMembershipDeleted(subscriber string, handler ComputeAllocationMembershipHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationMembershipDeleteEvent, handler)
}
diff --git a/pkg/events/compute_allocation_resource_mapping_subscribe.go b/pkg/events/compute_allocation_resource_mapping_subscribe.go
index d30261ace..3117712b0 100644
--- a/pkg/events/compute_allocation_resource_mapping_subscribe.go
+++ b/pkg/events/compute_allocation_resource_mapping_subscribe.go
@@ -19,48 +19,29 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// ComputeAllocationResourceMappingHandler handles compute allocation resource mapping lifecycle events with a typed payload.
-type ComputeAllocationResourceMappingHandler func(ctx context.Context, mapping models.ComputeAllocationResourceMapping)
+type ComputeAllocationResourceMappingHandler func(ctx context.Context, mapping models.ComputeAllocationResourceMapping) error
// SubscribeComputeAllocationResourceMappingCreated registers a typed handler invoked whenever a
// compute_allocation_resource_mapping::create event is published. Events with payloads that are
// not a models.ComputeAllocationResourceMapping (or *models.ComputeAllocationResourceMapping) are
// dropped with a warning log.
-func (b *Bus) SubscribeComputeAllocationResourceMappingCreated(handler ComputeAllocationResourceMappingHandler) {
- b.subscribeComputeAllocationResourceMapping(ComputeAllocationResourceMappingCreateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationResourceMappingCreated(subscriber string, handler ComputeAllocationResourceMappingHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationResourceMappingCreateEvent, handler)
}
// SubscribeComputeAllocationResourceMappingUpdated registers a typed handler invoked whenever a
// compute_allocation_resource_mapping::update event is published.
-func (b *Bus) SubscribeComputeAllocationResourceMappingUpdated(handler ComputeAllocationResourceMappingHandler) {
- b.subscribeComputeAllocationResourceMapping(ComputeAllocationResourceMappingUpdateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationResourceMappingUpdated(subscriber string, handler ComputeAllocationResourceMappingHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationResourceMappingUpdateEvent, handler)
}
// SubscribeComputeAllocationResourceMappingDeleted registers a typed handler invoked whenever a
// compute_allocation_resource_mapping::delete event is published.
-func (b *Bus) SubscribeComputeAllocationResourceMappingDeleted(handler ComputeAllocationResourceMappingHandler) {
- b.subscribeComputeAllocationResourceMapping(ComputeAllocationResourceMappingDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeComputeAllocationResourceMapping(topic EventType, handler ComputeAllocationResourceMappingHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch m := value.(type) {
- case models.ComputeAllocationResourceMapping:
- handler(ctx, m)
- case *models.ComputeAllocationResourceMapping:
- if m != nil {
- handler(ctx, *m)
- }
- default:
- slog.Warn("compute allocation resource mapping event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeComputeAllocationResourceMappingDeleted(subscriber string, handler ComputeAllocationResourceMappingHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationResourceMappingDeleteEvent, handler)
}
diff --git a/pkg/events/compute_allocation_resource_subscribe.go b/pkg/events/compute_allocation_resource_subscribe.go
index 8f91b19d8..26dafb800 100644
--- a/pkg/events/compute_allocation_resource_subscribe.go
+++ b/pkg/events/compute_allocation_resource_subscribe.go
@@ -19,48 +19,29 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// ComputeAllocationResourceHandler handles compute allocation resource lifecycle events with a typed payload.
-type ComputeAllocationResourceHandler func(ctx context.Context, resource models.ComputeAllocationResource)
+type ComputeAllocationResourceHandler func(ctx context.Context, resource models.ComputeAllocationResource) error
// SubscribeComputeAllocationResourceCreated registers a typed handler invoked whenever a
// compute_allocation_resource::create event is published. Events with payloads that are
// not a models.ComputeAllocationResource (or *models.ComputeAllocationResource) are dropped
// with a warning log.
-func (b *Bus) SubscribeComputeAllocationResourceCreated(handler ComputeAllocationResourceHandler) {
- b.subscribeComputeAllocationResource(ComputeAllocationResourceCreateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationResourceCreated(subscriber string, handler ComputeAllocationResourceHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationResourceCreateEvent, handler)
}
// SubscribeComputeAllocationResourceUpdated registers a typed handler invoked whenever a
// compute_allocation_resource::update event is published.
-func (b *Bus) SubscribeComputeAllocationResourceUpdated(handler ComputeAllocationResourceHandler) {
- b.subscribeComputeAllocationResource(ComputeAllocationResourceUpdateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationResourceUpdated(subscriber string, handler ComputeAllocationResourceHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationResourceUpdateEvent, handler)
}
// SubscribeComputeAllocationResourceDeleted registers a typed handler invoked whenever a
// compute_allocation_resource::delete event is published.
-func (b *Bus) SubscribeComputeAllocationResourceDeleted(handler ComputeAllocationResourceHandler) {
- b.subscribeComputeAllocationResource(ComputeAllocationResourceDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeComputeAllocationResource(topic EventType, handler ComputeAllocationResourceHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch r := value.(type) {
- case models.ComputeAllocationResource:
- handler(ctx, r)
- case *models.ComputeAllocationResource:
- if r != nil {
- handler(ctx, *r)
- }
- default:
- slog.Warn("compute allocation resource event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeComputeAllocationResourceDeleted(subscriber string, handler ComputeAllocationResourceHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationResourceDeleteEvent, handler)
}
diff --git a/pkg/events/compute_allocation_subscribe.go b/pkg/events/compute_allocation_subscribe.go
index 0ce1533eb..07a27b045 100644
--- a/pkg/events/compute_allocation_subscribe.go
+++ b/pkg/events/compute_allocation_subscribe.go
@@ -19,48 +19,29 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// ComputeAllocationHandler handles compute allocation lifecycle events with a typed payload.
-type ComputeAllocationHandler func(ctx context.Context, allocation models.ComputeAllocation)
+type ComputeAllocationHandler func(ctx context.Context, allocation models.ComputeAllocation) error
// SubscribeComputeAllocationCreated registers a typed handler invoked whenever a
// compute_allocation::create event is published. Events with payloads that are
// not a models.ComputeAllocation (or *models.ComputeAllocation) are dropped
// with a warning log.
-func (b *Bus) SubscribeComputeAllocationCreated(handler ComputeAllocationHandler) {
- b.subscribeComputeAllocation(ComputeAllocationCreateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationCreated(subscriber string, handler ComputeAllocationHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationCreateEvent, handler)
}
// SubscribeComputeAllocationUpdated registers a typed handler invoked whenever a
// compute_allocation::update event is published.
-func (b *Bus) SubscribeComputeAllocationUpdated(handler ComputeAllocationHandler) {
- b.subscribeComputeAllocation(ComputeAllocationUpdateEvent, handler)
+func (b *Bus) SubscribeComputeAllocationUpdated(subscriber string, handler ComputeAllocationHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationUpdateEvent, handler)
}
// SubscribeComputeAllocationDeleted registers a typed handler invoked whenever a
// compute_allocation::delete event is published.
-func (b *Bus) SubscribeComputeAllocationDeleted(handler ComputeAllocationHandler) {
- b.subscribeComputeAllocation(ComputeAllocationDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeComputeAllocation(topic EventType, handler ComputeAllocationHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch a := value.(type) {
- case models.ComputeAllocation:
- handler(ctx, a)
- case *models.ComputeAllocation:
- if a != nil {
- handler(ctx, *a)
- }
- default:
- slog.Warn("compute allocation event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeComputeAllocationDeleted(subscriber string, handler ComputeAllocationHandler) {
+ subscribeTyped(b, subscriber, ComputeAllocationDeleteEvent, handler)
}
diff --git a/pkg/events/compute_cluster_user_subscribe.go b/pkg/events/compute_cluster_user_subscribe.go
index 1ef7d60d1..6a4749d35 100644
--- a/pkg/events/compute_cluster_user_subscribe.go
+++ b/pkg/events/compute_cluster_user_subscribe.go
@@ -19,47 +19,28 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// ComputeClusterUserHandler handles compute-cluster user lifecycle events
// with a typed payload.
-type ComputeClusterUserHandler func(ctx context.Context, user models.ComputeClusterUser)
+type ComputeClusterUserHandler func(ctx context.Context, user models.ComputeClusterUser) error
// SubscribeComputeClusterUserCreated registers a typed handler invoked
// whenever a compute_cluster_user::create event is published.
-func (b *Bus) SubscribeComputeClusterUserCreated(handler ComputeClusterUserHandler) {
- b.subscribeComputeClusterUser(ComputeClusterUserCreateEvent, handler)
+func (b *Bus) SubscribeComputeClusterUserCreated(subscriber string, handler ComputeClusterUserHandler) {
+ subscribeTyped(b, subscriber, ComputeClusterUserCreateEvent, handler)
}
// SubscribeComputeClusterUserUpdated registers a typed handler invoked
// whenever a compute_cluster_user::update event is published.
-func (b *Bus) SubscribeComputeClusterUserUpdated(handler ComputeClusterUserHandler) {
- b.subscribeComputeClusterUser(ComputeClusterUserUpdateEvent, handler)
+func (b *Bus) SubscribeComputeClusterUserUpdated(subscriber string, handler ComputeClusterUserHandler) {
+ subscribeTyped(b, subscriber, ComputeClusterUserUpdateEvent, handler)
}
// SubscribeComputeClusterUserDeleted registers a typed handler invoked
// whenever a compute_cluster_user::delete event is published.
-func (b *Bus) SubscribeComputeClusterUserDeleted(handler ComputeClusterUserHandler) {
- b.subscribeComputeClusterUser(ComputeClusterUserDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeComputeClusterUser(topic EventType, handler ComputeClusterUserHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch u := value.(type) {
- case models.ComputeClusterUser:
- handler(ctx, u)
- case *models.ComputeClusterUser:
- if u != nil {
- handler(ctx, *u)
- }
- default:
- slog.Warn("compute cluster user event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeComputeClusterUserDeleted(subscriber string, handler ComputeClusterUserHandler) {
+ subscribeTyped(b, subscriber, ComputeClusterUserDeleteEvent, handler)
}
diff --git a/pkg/events/organization_subscribe.go b/pkg/events/organization_subscribe.go
index 1a1a94f21..87ab0d876 100644
--- a/pkg/events/organization_subscribe.go
+++ b/pkg/events/organization_subscribe.go
@@ -19,47 +19,28 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// OrganizationHandler handles organization lifecycle events with a typed payload.
-type OrganizationHandler func(ctx context.Context, organization models.Organization)
+type OrganizationHandler func(ctx context.Context, organization models.Organization) error
// SubscribeOrganizationCreated registers a typed handler invoked whenever an
// organization::create event is published. Events with payloads that are not a
// models.Organization (or *models.Organization) are dropped with a warning log.
-func (b *Bus) SubscribeOrganizationCreated(handler OrganizationHandler) {
- b.subscribeOrganization(OrganizationCreateEvent, handler)
+func (b *Bus) SubscribeOrganizationCreated(subscriber string, handler OrganizationHandler) {
+ subscribeTyped(b, subscriber, OrganizationCreateEvent, handler)
}
// SubscribeOrganizationUpdated registers a typed handler invoked whenever an
// organization::update event is published.
-func (b *Bus) SubscribeOrganizationUpdated(handler OrganizationHandler) {
- b.subscribeOrganization(OrganizationUpdateEvent, handler)
+func (b *Bus) SubscribeOrganizationUpdated(subscriber string, handler OrganizationHandler) {
+ subscribeTyped(b, subscriber, OrganizationUpdateEvent, handler)
}
// SubscribeOrganizationDeleted registers a typed handler invoked whenever an
// organization::delete event is published.
-func (b *Bus) SubscribeOrganizationDeleted(handler OrganizationHandler) {
- b.subscribeOrganization(OrganizationDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeOrganization(topic EventType, handler OrganizationHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch o := value.(type) {
- case models.Organization:
- handler(ctx, o)
- case *models.Organization:
- if o != nil {
- handler(ctx, *o)
- }
- default:
- slog.Warn("organization event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeOrganizationDeleted(subscriber string, handler OrganizationHandler) {
+ subscribeTyped(b, subscriber, OrganizationDeleteEvent, handler)
}
diff --git a/pkg/events/project_subscribe.go b/pkg/events/project_subscribe.go
index 26581bf73..06653113a 100644
--- a/pkg/events/project_subscribe.go
+++ b/pkg/events/project_subscribe.go
@@ -19,49 +19,30 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// ProjectHandler handles project lifecycle events with a typed payload.
-type ProjectHandler func(ctx context.Context, project models.Project)
+type ProjectHandler func(ctx context.Context, project models.Project) error
// SubscribeProjectCreated registers a typed handler invoked whenever a
// project::create event is published. Events with payloads that are not a
// models.Project (or *models.Project) are dropped with a warning log.
-func (b *Bus) SubscribeProjectCreated(handler ProjectHandler) {
- b.subscribeProject(ProjectCreateEvent, handler)
+func (b *Bus) SubscribeProjectCreated(subscriber string, handler ProjectHandler) {
+ subscribeTyped(b, subscriber, ProjectCreateEvent, handler)
}
// SubscribeProjectUpdated registers a typed handler invoked whenever a
// project::update event is published. Events with payloads that are not a
// models.Project (or *models.Project) are dropped with a warning log.
-func (b *Bus) SubscribeProjectUpdated(handler ProjectHandler) {
- b.subscribeProject(ProjectUpdateEvent, handler)
+func (b *Bus) SubscribeProjectUpdated(subscriber string, handler ProjectHandler) {
+ subscribeTyped(b, subscriber, ProjectUpdateEvent, handler)
}
// SubscribeProjectDeleted registers a typed handler invoked whenever a
// project::delete event is published. Events with payloads that are not a
// models.Project (or *models.Project) are dropped with a warning log.
-func (b *Bus) SubscribeProjectDeleted(handler ProjectHandler) {
- b.subscribeProject(ProjectDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeProject(topic EventType, handler ProjectHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch p := value.(type) {
- case models.Project:
- handler(ctx, p)
- case *models.Project:
- if p != nil {
- handler(ctx, *p)
- }
- default:
- slog.Warn("project event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeProjectDeleted(subscriber string, handler ProjectHandler) {
+ subscribeTyped(b, subscriber, ProjectDeleteEvent, handler)
}
diff --git a/pkg/events/types.go b/pkg/events/types.go
index 4d9cb2b62..999ee16c7 100644
--- a/pkg/events/types.go
+++ b/pkg/events/types.go
@@ -127,11 +127,18 @@ type Event struct {
}
// EventSubscriberFunc is a function type that can be registered to receive events from the bus.
-type EventSubscriberFunc func(ctx context.Context, event Event, value interface{})
+// A returned error means the event was not handled.
+type EventSubscriberFunc func(ctx context.Context, event Event, value interface{}) error
+
+// subscription is one named subscriber's handler for a topic.
+type subscription struct {
+ subscriber string
+ handler EventSubscriberFunc
+}
// Bus is a lightweight, in-memory, topic-based pub/sub event bus.
// Modules publish and subscribe by topic without knowing about each other.
type Bus struct {
mu sync.RWMutex
- subs map[string][]EventSubscriberFunc
+ subs map[string][]subscription
}
diff --git a/pkg/events/user_identity_subscribe.go b/pkg/events/user_identity_subscribe.go
index 06159522a..7d4ceba40 100644
--- a/pkg/events/user_identity_subscribe.go
+++ b/pkg/events/user_identity_subscribe.go
@@ -19,46 +19,27 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// UserIdentityHandler handles user-identity lifecycle events with a typed payload.
-type UserIdentityHandler func(ctx context.Context, identity models.UserIdentity)
+type UserIdentityHandler func(ctx context.Context, identity models.UserIdentity) error
// SubscribeUserIdentityCreated registers a typed handler invoked whenever a
// user_identity::create event is published.
-func (b *Bus) SubscribeUserIdentityCreated(handler UserIdentityHandler) {
- b.subscribeUserIdentity(UserIdentityCreateEvent, handler)
+func (b *Bus) SubscribeUserIdentityCreated(subscriber string, handler UserIdentityHandler) {
+ subscribeTyped(b, subscriber, UserIdentityCreateEvent, handler)
}
// SubscribeUserIdentityUpdated registers a typed handler invoked whenever a
// user_identity::update event is published.
-func (b *Bus) SubscribeUserIdentityUpdated(handler UserIdentityHandler) {
- b.subscribeUserIdentity(UserIdentityUpdateEvent, handler)
+func (b *Bus) SubscribeUserIdentityUpdated(subscriber string, handler UserIdentityHandler) {
+ subscribeTyped(b, subscriber, UserIdentityUpdateEvent, handler)
}
// SubscribeUserIdentityDeleted registers a typed handler invoked whenever a
// user_identity::delete event is published.
-func (b *Bus) SubscribeUserIdentityDeleted(handler UserIdentityHandler) {
- b.subscribeUserIdentity(UserIdentityDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeUserIdentity(topic EventType, handler UserIdentityHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch e := value.(type) {
- case models.UserIdentity:
- handler(ctx, e)
- case *models.UserIdentity:
- if e != nil {
- handler(ctx, *e)
- }
- default:
- slog.Warn("user identity event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeUserIdentityDeleted(subscriber string, handler UserIdentityHandler) {
+ subscribeTyped(b, subscriber, UserIdentityDeleteEvent, handler)
}
diff --git a/pkg/events/user_subscribe.go b/pkg/events/user_subscribe.go
index 794bbd3a8..be5c9bb81 100644
--- a/pkg/events/user_subscribe.go
+++ b/pkg/events/user_subscribe.go
@@ -19,47 +19,28 @@ package events
import (
"context"
- "log/slog"
"github.com/apache/airavata-custos/pkg/models"
)
// UserHandler handles user lifecycle events with a typed payload.
-type UserHandler func(ctx context.Context, user models.User)
+type UserHandler func(ctx context.Context, user models.User) error
// SubscribeUserCreated registers a typed handler invoked whenever a
// user::create event is published. Events with payloads that are not a
// models.User (or *models.User) are dropped with a warning log.
-func (b *Bus) SubscribeUserCreated(handler UserHandler) {
- b.subscribeUser(UserCreateEvent, handler)
+func (b *Bus) SubscribeUserCreated(subscriber string, handler UserHandler) {
+ subscribeTyped(b, subscriber, UserCreateEvent, handler)
}
// SubscribeUserUpdated registers a typed handler invoked whenever a
// user::update event is published.
-func (b *Bus) SubscribeUserUpdated(handler UserHandler) {
- b.subscribeUser(UserUpdateEvent, handler)
+func (b *Bus) SubscribeUserUpdated(subscriber string, handler UserHandler) {
+ subscribeTyped(b, subscriber, UserUpdateEvent, handler)
}
// SubscribeUserDeleted registers a typed handler invoked whenever a
// user::delete event is published.
-func (b *Bus) SubscribeUserDeleted(handler UserHandler) {
- b.subscribeUser(UserDeleteEvent, handler)
-}
-
-func (b *Bus) subscribeUser(topic EventType, handler UserHandler) {
- b.Subscribe(topic, func(ctx context.Context, event Event, value interface{}) {
- switch u := value.(type) {
- case models.User:
- handler(ctx, u)
- case *models.User:
- if u != nil {
- handler(ctx, *u)
- }
- default:
- slog.Warn("user event payload has unexpected type",
- "type", event.Type,
- "got", value,
- )
- }
- })
+func (b *Bus) SubscribeUserDeleted(subscriber string, handler UserHandler) {
+ subscribeTyped(b, subscriber, UserDeleteEvent, handler)
}
diff --git a/pkg/service/identity_resolver_integration_test.go b/pkg/service/identity_resolver_integration_test.go
index 9eac0b91a..64565bbf6 100644
--- a/pkg/service/identity_resolver_integration_test.go
+++ b/pkg/service/identity_resolver_integration_test.go
@@ -262,8 +262,9 @@ func TestResolveCaller_EmailFallback_PublishesIdentityCreated(t *testing.T) {
user := seedPendingUserWithEmail(t, database, "provision-first@example.edu")
got := make(chan models.UserIdentity, 1)
- bus.SubscribeUserIdentityCreated(func(_ context.Context, identity models.UserIdentity) {
+ bus.SubscribeUserIdentityCreated("test-subscriber", func(_ context.Context, identity models.UserIdentity) error {
got <- identity
+ return nil
})
if _, _, err := svc.ResolveCaller(ctx(), &identity.Claims{