Notifications Services Configuration#

Goal: Deploy optional Notifications Services (RabbitMQ, event aggregation, event consumer) so the Storage Service can publish file events and clients can consume them.

When to use: Enable when you need event-driven workflows (e.g., Kit-based apps or other services reacting to file create/delete/modify). If you do not need notifications, skip this section and continue with the Discovery Service.

The implementation uses RabbitMQ as the event broker with two components: event aggregation (publishing) and event consumer. You deploy and enable them from the Storage Service or other services that publish or consume events.

Note

The ports for RabbitMQ are only ever used within the Kubernetes cluster. They are not exposed to the outside world.

Service Ports#

Service	Port	Description
RabbitMQ	5672	AMQP messaging protocol
RabbitMQ Management	15672	Management UI (if enabled)
Event Aggregation	50051	gRPC endpoint for publishing events
Event Consumer	8000	HTTP/REST endpoint for consuming events (`/api/v1beta/events/stream`)

Pulling the RabbitMQ Helm chart#

For our notifications services, we use RabbitMQ as the event broker. We will need to pull the RabbitMQ Helm chart from the NGC catalog. Replace {NGC_API_KEY} with your NGC API key.

# pull the chart from NGC
helm fetch https://helm.ngc.nvidia.com/nvidia/omniverse/charts/rabbitmq-99.3.0.tgz --username='$oauthtoken' --password={NGC_API_KEY}

# unpack the chart and cd into the directory
tar -xvf rabbitmq-99.3.0.tgz
cd rabbitmq

Configuring the RabbitMQ Helm chart#

We need to set values for the RabbitMQ Helm chart to use our event broker backend. We will use rabbitmq-values.yaml as the name of our values file.

# create and edit the local values file, we're using VS Code for this example
code rabbitmq-values.yaml

Pull Secrets#

Set your rabbitmq-values.yaml to use the ngcpull-secret for image pulling (same secret as for the Storage Service). You will also need to point it to the correct NGC registry.

image:
  pullSecrets:
    - name: ngcpull-secret
  registry: nvcr.io
  repositoryPrefix: nvidia/omniverse
  repository: rabbitmq
  tag: 4.1.3-debian-12-r1

rabbitmq-values.yaml

Configure RabbitMQ Password#

We need to configure the RabbitMQ with a password that is stored in a Kubernetes secret. Replace {RABBITMQ_PASSWORD} with the password you want to use.

# create a Kubernetes secret with the password
kubectl create secret generic rabbitmq-service-env --from-literal=rabbitmq-password={RABBITMQ_PASSWORD} --namespace storage-apis

Now you need to make sure to use the secret in your rabbitmq-values.yaml file.

auth:
  username: rabbitmq-user
  existingPasswordSecret: "rabbitmq-service-env" # this is the secret name we created above
  existingSecretPasswordKey: "rabbitmq-password" # this is the key in the secret with the password

rabbitmq-values.yaml

As this is a minimal deployment, we will avoid using persistent storage for the RabbitMQ cluster that would normally be used for durable queues.

Note

Persistence is disabled to keep the system stateless. If you need persistence for your deployment, enable and configure your needed storage class.

persistence:
  enabled: false

Configure RabbitMQ VHosts#

We use RabbitMQ virtual hosts (vhosts) to isolate the notifications services from other services that may share the same RabbitMQ instance. This ensures that queues, exchanges, and permissions are scoped to the notifications vhost and won’t conflict with other workloads.

Add the following to your rabbitmq-values.yaml to configure RabbitMQ to create the notifications vhost on startup. This uses loadDefinition to load a JSON definition from a Kubernetes secret that is created by the chart via extraSecrets and an initScript to fix the password for the vhost:

# Load vhost definitions on startup
loadDefinition:
  enabled: true
  existingSecret: rabbitmq-load-definition
extraSecrets:
  rabbitmq-load-definition:
    load_definition.json: |
      {
        "vhosts": [
          {"name": "/"},
          {"name": "notifications"}
        ]
      }
initScripts:
  setup-permissions.sh: |
    #!/bin/bash
    rabbitmqctl set_permissions -p notifications rabbitmq-user ".*" ".*" ".*"

rabbitmq-values.yaml

Note

The user field in the permissions must match the auth.username value you configured above (rabbitmq-user). The default / vhost is included to retain standard RabbitMQ functionality, and the notifications vhost is where the event aggregation and event consumer services will operate.

RabbitMQ Notification Connection Secret#

There is an additional Kubernetes secret you will need to create for the Notification Services for them to connect to the RabbitMQ service. We will use this later on when deploying each the Event Aggregation and Event Consumer Services. This is used to store the RabbitMQ amqp uri and is formatted as amqp://{username}:{RABBITMQ_PASSWORD}@{rabbitmq-service-name}:{port}/notifications. Note the /notifications path at the end, which directs the services to use the notifications vhost we configured above.

# create a Kubernetes secret with the RabbitMQ URI (note the /notifications vhost path)
kubectl create secret generic event-services-env --from-literal=rabbitmq_uri=amqp://rabbitmq-user:{RABBITMQ_PASSWORD}@rabbitmq:5672/notifications --namespace storage-apis

Deploying the RabbitMQ Helm chart#

Now you can deploy the RabbitMQ Helm chart using the helm install command.

# deploy the chart
helm upgrade --install rabbitmq -n storage-apis -f rabbitmq-values.yaml .

Validate the RabbitMQ Deployment#

You can validate the RabbitMQ deployment by checking the pods are running and the service is accessible.

# check the pods are running
kubectl get pods -n storage-apis

# check the service is accessible
kubectl get service -n storage-apis

Pulling the Aggregation Services Helm charts#

With RabbitMQ and our backend storage configured, we will need to pull and unpack the Notifications Aggregation Service Helm Chart from the NGC catalog. Replace {NGC_API_KEY} with your NGC API key.

# pull the chart from NGC
helm fetch https://helm.ngc.nvidia.com/nvidia/omniverse/charts/event-aggregation-service-1.5.52.tgz --username='$oauthtoken' --password={NGC_API_KEY}

# unpack the chart and cd into the directory
tar -xvf event-aggregation-service-1.5.52.tgz
cd event-aggregation-service

Configuring the Aggregation Service#

Create your event-aggregation-values.yaml file for the Aggregation Service.

# create and edit the local values file, we're using VS Code for this example
code event-aggregation-values.yaml

Pull Secrets#

Set your event-aggregation-values.yaml to use the ngcpull-secret for image pulling (same secret as for the Storage Service).

image:
  pullSecrets:
    - name: ngcpull-secret

event-aggregation-values.yaml

Configure Backend Access#

We need to configure the Aggregation Service to access your backend event broker. The chart defaults to using a secret named event-aggregation-env, but since we created a shared secret named event-services-env above, we need to override it:

# Override the default secret name (event-aggregation-env) to use our shared secret
envSecretName: event-services-env

event-aggregation-values.yaml

Install the Aggregation Service#

Now you can install the Aggregation Service using the helm install command.

# install the chart
helm upgrade --install event-aggregation-service -n storage-apis -f event-aggregation-values.yaml .

Validate the Aggregation Service Deployment#

You can validate the Aggregation Service deployment by checking the pods are running and the service is accessible.

# check the pods are running
kubectl get pods -n storage-apis

# check the service is accessible
kubectl get service -n storage-apis

Pulling the Consumer Services Helm charts#

We will now pull and unpack the Notifications Consumer Service Helm Chart from the NGC catalog. Replace {NGC_API_KEY} with your NGC API key. This is the service that external clients will use to consume events from the Notifications Services.

helm fetch https://helm.ngc.nvidia.com/nvidia/omniverse/charts/event-consumer-service-1.9.6.tgz --username='$oauthtoken' --password={NGC_API_KEY}

# unpack the chart and cd into the directory
tar -xvf event-consumer-service-1.9.6.tgz
cd event-consumer-service

Configuring the Consumer Service#

Create your event-consumer-values.yaml file for the Consumer Service.

# create and edit the local values file, we're using VS Code for this example
code event-consumer-values.yaml

Pull Secrets#

Set your event-consumer-values.yaml to use the ngcpull-secret for image pulling (same secret as for the Storage Service).

image:
  pullSecrets:
    - name: ngcpull-secret

event-consumer-values.yaml

Configure RabbitMQ Connection#

We need to configure the Consumer Service to connect to the RabbitMQ service. Similar to the Aggregation Service, the chart defaults to using a secret named event-consumer-env, but since we created a shared secret named event-services-env above, we need to override it:

# Override the default secret name (event-consumer-env) to use our shared secret
envSecretName: event-services-env

event-consumer-values.yaml Install the Consumer Service ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Now you can install the Consumer Service using the helm install command.

# install the chart
helm upgrade --install event-consumer-service -n storage-apis -f event-consumer-values.yaml .

Validate the Consumer Service Deployment#

You can validate the Consumer Service deployment by checking the pods are running and the service is accessible.

# check the pods are running
kubectl get pods -n storage-apis

# check the service is accessible
kubectl get service -n storage-apis

Configuration on your Event Broker Backend#

Next, configure your backend storage Event Broker to work with the Notifications Services. This will require updates to your storage-values.yaml file as you integrate the Notifications Services into your storage service.

Setting up the Notification Service client credentials#

Issue client credentials for the Storage Service application so it can publish events via the Notification Service.

Setup AWS S3 event subscription#

Single region, single consumer#

Use this when you have one or more S3 buckets in the same AWS region and the Storage Service is the only consumer of these events.

Setting up an SQS queue#

In the AWS Console, open the SQS Dashboard and click Create queue.
Set Type → Standard, Name, and Message retention period (e.g. several hours).

For Access policy choose Advanced and add a statement so S3 can send messages to the queue. In the Statement array, add an object like:

{
  "Sid": "<statement-id>",
  "Effect": "Allow",
  "Principal": { "Service": "s3.amazonaws.com" },
  "Action": "SQS:SendMessage",
  "Resource": "arn:aws:sqs:<aws-region>:<aws-account-id>:<sqs-queue-name>",
  "Condition": {
    "StringEquals": { "aws:SourceAccount": "<aws-account-id>" },
    "ArnLike": { "aws:SourceArn": "arn:aws:s3:*:*:<s3-bucket-name>" }
  }
}

Replace <statement-id>, <aws-region>, <aws-account-id>, <sqs-queue-name>, and <s3-bucket-name> with your values. The full policy should include both the default owner statement and this S3 statement.

Create the queue and note its URL.

Setting up an Event Subscription (S3 → SQS)#

In the AWS Console open S3, select the bucket, then Properties → Event notifications.
Click Create event notification. Set Event name; Event types → All object create events and All object removal events; Destination → SQS queue and select the queue from the previous step.
Save changes.

Many regions or multiple consumers#

Use this when you have multiple event consumers, multiple queues, S3 buckets in different regions, or multiple Storage Service deployments.

Setting up an SQS queue (multi-region/multi-consumer)#

Create a Standard SQS queue (name, message retention as above).

Access policy: - Option A (Recommended): Use Basic; AWS will update the queue policy when you create the SNS subscription. - Option B: Use Advanced and add a statement allowing SNS to send messages:

{
  "Sid": "AllowSNSPublish",
  "Effect": "Allow",
  "Principal": { "Service": "sns.amazonaws.com" },
  "Action": "SQS:SendMessage",
  "Resource": "arn:aws:sqs:<aws-region>:<aws-account-id>:<sqs-queue-name>",
  "Condition": {
    "ArnEquals": { "aws:SourceArn": "arn:aws:sns:<aws-region>:<aws-account-id>:<sns-topic-name>" }
  }
}

Replace placeholders with your values.

Create the queue and note its ARN.

Note

Option A is safe because AWS sets the correct permissions when you create the SNS subscription. Option B gives explicit control.

Setting up an Event Subscription (S3 → SNS)#

Repeat per S3 bucket:

S3 → select bucket → Properties → Event notifications → Create event notification.
Set Event name; Event types → All object create events and All object removal events; Destination → SNS topic → Enter SNS topic ARN and paste the SNS topic ARN for this region.
Save changes.

Validation (AWS)#

AWS sends a test notification per event subscription. To verify:

SQS Dashboard → open your queue → Send and receive messages.
Receive Messages → Edit poll settings → set Polling duration to 1 second → Poll for messages.
Inspect messages; they should look like {"Service":"Amazon S3","Event":"s3:TestEvent",...}. If you have many subscriptions, delete seen messages and poll again.

Setting up Service Credentials (AWS SQS)#

Note

Unified identity: If the Storage Service already uses an IAM role (e.g. IRSA on EKS) for S3 bucket access, attach the SQS permissions below to that same role instead of creating a separate user. One identity then has both bucket and notification access; you do not need access keys. Configure the chart to use the pod’s service account (omit accessKeyId and secretAccessKey in values). If you use a separate IAM user for SQS, create the policy and user as in Option 1 and inject the keys as shown in Setting up the Service.

Option 1: Custom policy (Recommended when not using a service account)

IAM → Policies → Create policy → JSON and use:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "StorageServiceSQSAccess",
      "Effect": "Allow",
      "Action": [
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:GetQueueAttributes",
        "sqs:GetQueueUrl"
      ],
      "Resource": "arn:aws:sqs:<aws-region>:<aws-account-id>:<sqs-queue-name>"
    }
  ]
}

Replace <aws-region>, <aws-account-id>, <sqs-queue-name>. Name the policy (e.g. StorageServiceSQSConsumerPolicy) and create it.

Either attach this policy to the same IAM role used for S3 (e.g. your IRSA role) so bucket and SQS permissions are unified, or IAM → Users → Create user → attach the policy → create user.
If you created a separate user: open the user → Security credentials → Create access key → copy Access key and Secret access key. If you attached the policy to an existing role (e.g. IRSA), you do not need access keys.

Option 2: For quick testing only, you can use the managed policy AmazonSQSFullAccess; it is not recommended for production (least privilege).

Setup Azure Blob event subscription#

Setting up a Storage Event Subscription#

Log into the Azure Portal.
Create or select a Service Bus Namespace via the Service Bus browser. When creating a Namespace, place it in the same region as your other services. For the event subscription, the Basic tier is sufficient.
In the Service Bus Namespace view click + Queue.
Set Name and Message time to live (e.g. several hours instead of the default 14 days). Review and create the queue.
In the Storage Accounts browser, open the Storage Account for which you want the event subscription.
In the left panel, open Events and click + Event Subscription.
Configure: Name; Event Schema → Cloud Event Schema v1.0; Filter to Event Types → ensure Blob Created and Blob Deleted are selected; Endpoint Type → Service Bus Queue and configure the endpoint to the queue from step 4.
Create the event subscription.

Configure Notifications on the Storage Service#

The following environment variables must be available to the Storage Service containers, except where you use a single identity (see below). You can inject them via a Kubernetes secret and extraEnvs (as with bucket credentials), or via another mechanism such as Vault. You only need to add the environment variables for the Notifications Services you are using.

AZURE_SERVICE_BUS_POLICY_KEY — Azure Service Bus: open your namespace → Shared access policy → RootManageSharedAccessKey → copy Primary key. Omit if not using Azure Service Bus.
SQS_ACCESS_KEY_ID and SQS_SECRET_ACCESS_KEY — only if you use a separate IAM user for SQS. When the Storage Service uses an IAM role (e.g. IRSA) for S3, attach SQS permissions to that same role and do not set these; the pod will use its service account and you omit them from the secret and from extraEnvs below.
NOTIFICATION_SERVICE_CLIENT_SECRET — from the Notification Service client credentials you created.

Create a Kubernetes secret with the values (replace placeholders and namespace; omit SQS keys when using IRSA):

kubectl create secret generic storage-notifications-secret \
  --from-literal=AZURE_SERVICE_BUS_POLICY_KEY="<service-bus-policy-key>" \
  --from-literal=SQS_ACCESS_KEY_ID="<aws-access-key-id>" \
  --from-literal=SQS_SECRET_ACCESS_KEY="<aws-secret-access-key>" \
  --from-literal=NOTIFICATION_SERVICE_CLIENT_SECRET="<client-secret>" \
  --namespace storage-apis

Add extraEnvs to your storage-values.yaml file so the Storage Service gets these environment variables from the secret at deploy time:

extraEnvs:
  - name: AZURE_SERVICE_BUS_POLICY_KEY
    valueFrom:
      secretKeyRef:
        name: storage-notifications-secret
        key: AZURE_SERVICE_BUS_POLICY_KEY
  - name: SQS_ACCESS_KEY_ID
    valueFrom:
      secretKeyRef:
        name: storage-notifications-secret
        key: SQS_ACCESS_KEY_ID
  - name: SQS_SECRET_ACCESS_KEY
    valueFrom:
      secretKeyRef:
        name: storage-notifications-secret
        key: SQS_SECRET_ACCESS_KEY
  - name: NOTIFICATION_SERVICE_CLIENT_SECRET
    valueFrom:
      secretKeyRef:
        name: storage-notifications-secret
        key: NOTIFICATION_SERVICE_CLIENT_SECRET

storage-values.yaml

When using IRSA for SQS, omit the SQS_ACCESS_KEY_ID and SQS_SECRET_ACCESS_KEY entries from the secret and from extraEnvs. If you already have extraEnvs (e.g. for bucket credentials), append these items to the same list.

Update the storage-values.yaml file to enable storage events and the notification client. Make sure to replace the placeholders with the values you obtained from the previous steps.

Replace placeholders: <idp-tenant-id>, <client-id>, <scope> from Notification Service client setup; <service-bus-namespace-name>, <service-bus-queue-name> from Azure; <sqs-queue-url>, <aws-region> from AWS; <event-aggregation-svc-name>, <event-aggregation-svc-namespace> for your Notification Service deployment.

config:
  # ...
  storageEvents:
    azureServiceBus:
      enabled: true
      queueNamespace: "<service-bus-namespace-name>"
      queueName: "<service-bus-queue-name>"
      credentials:
        sharedAccessPolicyKey:
          policyName: "RootManageSharedAccessKey"
          policyKey: "${AZURE_SERVICE_BUS_POLICY_KEY}"

    sqs:
      enabled: true
      queueUrl: "<sqs-queue-url>"
      region: "<aws-region>"
      # When using IRSA/service account, omit accessKeyId and secretAccessKey
      credentials:
        accessKey:
          accessKeyId: "${SQS_ACCESS_KEY_ID}"
          secretAccessKey: "${SQS_SECRET_ACCESS_KEY}"

  notificationClient:
    enabled: true
    endpointUrl: "http://<event-aggregation-svc-name>.<event-aggregation-svc-namespace>.svc.cluster.local:50051"
    secure: false

storage-values.yaml

Redeploy the Storage Service#

After including the above in your storage-values.yaml file, you can update the Storage Service using the helm upgrade command.

# install the chart
helm upgrade --install storage-service -n storage-apis -f storage-values.yaml .

Validate the Storage Service Deployment#

You can validate the Storage Service deployment by checking the pods are running and the service is accessible.

# check the pods are running
kubectl get pods -n storage-apis

# check the service is accessible
kubectl get service -n storage-apis

Testing the Notifications Services#

Now you can test the Notifications Services by subscribing to events and then triggering a storage change. This uses a non-durable subscription which is ideal for testing - it receives events in real-time while you’re connected.

You’ll need two terminal sessions: one to listen for events and one to trigger storage changes.

Note

The curl commands below use cluster-internal DNS names (e.g., event-consumer-service.storage-apis.svc.cluster.local), which are only accessible from within the Kubernetes cluster. To test from your local machine, use kubectl port-forward to make the services accessible on localhost. Each port-forward needs its own terminal session:

# port-forward the event consumer service (use deployment/ instead of service/
# because the chart also deploys an autoscaler service that can interfere)
kubectl port-forward -n storage-apis deployment/event-consumer-service 8000:8000

# port-forward the storage service (if not already forwarded from earlier testing)
kubectl port-forward -n storage-apis service/storage-service 8012:8012

When using port-forwarding, replace the cluster-internal hostnames in the curl commands below with localhost, e.g.:

http://event-consumer-service.storage-apis.svc.cluster.local:8000/... becomes http://localhost:8000/...
http://storage-service.storage-apis.svc.cluster.local:8012/... becomes http://localhost:8012/...

Trigger a Storage Event (Terminal 2)#

In a second terminal, create a file in your backend storage to trigger an event. Replace <YOUR BUCKET NAME> and <YOUR REGION> with your bucket name and region. (If you are not using AWS S3, you’ll need to modify the URL to match your backend storage.)

# Create a file to trigger an omni.storage.created event
export BUCKET_NAME=<YOUR BUCKET NAME>
export REGION=<YOUR REGION>
curl -X PUT "http://storage-service.storage-apis.svc.cluster.local:8012/v1beta/fileobject/by-address/https%3A%2F%2F${BUCKET_NAME}.s3.${REGION}.amazonaws.com%2Fhello.txt?data_object_size=20" \
  -H "Content-Type: application/octet-stream" \
  --data "Hello from Storage API!"

Verify Event Reception#

Back in Terminal 1, you should see an event appear similar to:

event: message
id: 12345
data: {"event_type":"omni.storage.created","principal_identity":"...","occurred_at":"...","published_at":"...","message":{"file_name":"hello.txt",...}}

event: reconnect_token
data: {"reconnect_token":"abc123xyz"}

If you see the event, your Notifications Services are working correctly. You can also test delete events by removing the file through the Storage Service.

Smoke Tests#

Both the Event Aggregation and Event Consumer services include smoke tests that verify their endpoints after deployment.

Event Aggregation Service — 4 tests (2 REST + 2 gRPC) that verify publishing single and batch events.

Event Consumer Service — 6 tests (3 REST + 3 gRPC) that verify durable queue lifecycle (create, update, delete) and non-durable stream reconnect token.

When tests.enabled=true, the tests run automatically on every helm install and helm upgrade, and can also be triggered on demand with helm test.

To enable and run smoke tests for each service:

# Event Aggregation
helm upgrade --install event-aggregation-service . -f event-aggregation-values.yaml \
  --set tests.enabled=true
helm test event-aggregation-service -n storage-apis

# Event Consumer
helm upgrade --install event-consumer-service . -f event-consumer-values.yaml \
  --set tests.enabled=true
helm test event-consumer-service -n storage-apis

Note

When authentication is enabled on the service, configure tests.auth in your values file so the test pod can obtain a Bearer token. The auth block reads credentials from a Kubernetes secret:

tests:
  enabled: true
  auth:
    microsoftUrl:
      valueFrom:
        secretKeyRef:
          name: smoke-test-auth
          key: microsoft-url
    credentials:
      valueFrom:
        secretKeyRef:
          name: smoke-test-auth
          key: credentials
    scope:
      valueFrom:
        secretKeyRef:
          name: smoke-test-auth
          key: scope

Create the secret before deploying:

kubectl create secret generic smoke-test-auth \
  --from-literal=microsoft-url="https://login.microsoftonline.com/{tenant-id}/oauth2/v2.0/token" \
  --from-literal=credentials="{client-id}:{client-secret}" \
  --from-literal=scope="{your-scope}" \
  --namespace storage-apis

event-aggregation-values.yaml

event-consumer-values.yaml

Notifications Services Configuration#

Service Ports#

Pulling the RabbitMQ Helm chart#

Configuring the RabbitMQ Helm chart#

Pull Secrets#

Configure RabbitMQ Password#

Configure RabbitMQ VHosts#

RabbitMQ Notification Connection Secret#

Deploying the RabbitMQ Helm chart#

Validate the RabbitMQ Deployment#

Pulling the Aggregation Services Helm charts#

Configuring the Aggregation Service#

Pull Secrets#

Configure Backend Access#

Install the Aggregation Service#

Validate the Aggregation Service Deployment#

Pulling the Consumer Services Helm charts#

Configuring the Consumer Service#

Pull Secrets#

Configure RabbitMQ Connection#

Validate the Consumer Service Deployment#

Configuration on your Event Broker Backend#

Setting up the Notification Service client credentials#

Setup AWS S3 event subscription#

Single region, single consumer#

Setting up an SQS queue#

Setting up an Event Subscription (S3 → SQS)#

Many regions or multiple consumers#

Setting up an SQS queue (multi-region/multi-consumer)#

Setting up an SNS topic#

Setting up an Event Subscription (S3 → SNS)#

Validation (AWS)#

Setting up Service Credentials (AWS SQS)#

Setup Azure Blob event subscription#

Setting up a Storage Event Subscription#

Configure Notifications on the Storage Service#

Redeploy the Storage Service#

Validate the Storage Service Deployment#

Testing the Notifications Services#

Subscribe to Events (Terminal 1)#

Trigger a Storage Event (Terminal 2)#

Verify Event Reception#

Smoke Tests#