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.

Warning

This minimal/testing path is fine for the non-durable streaming getting-started flow shown here, but the Event Consumer Service’s durable queues require persistence enabled (persistence.enabled: true) — with persistence disabled, durable queues and their messages are lost on pod restart. For high availability, durable queues also need a multi-node cluster (e.g. replicaCount: 3 + clustering.enabled: true). For production / durable configuration, see Event Consumer Dead-Letter Capture.

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.6.tgz --username='$oauthtoken' --password={NGC_API_KEY}

# unpack the chart and cd into the directory
tar -xvf event-aggregation-service-1.5.6.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

Note

Authentication for this getting-started flow. By default the service uses principalSource: jwtUnverified, which requires a bearer token on every request (including the Storage Service’s event publishes). For this tokenless walkthrough, set principalSource: disabled so the service runs without authentication. To configure real authentication and permissions instead, see Authentication and Permissions below.

# Tokenless getting-started: no auth middleware, no token required
principalSource: disabled

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.10.0.tgz --username='$oauthtoken' --password={NGC_API_KEY}

# unpack the chart and cd into the directory
tar -xvf event-consumer-service-1.10.0.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

Note

Authentication for this getting-started flow. As with the Aggregation Service, the Consumer Service defaults to principalSource: jwtUnverified and requires a bearer token on every request — including the tokenless event stream used in the testing section below. For this walkthrough, set principalSource: disabled. See Authentication and Permissions below to configure real authentication.

# Tokenless getting-started: no auth middleware, no token required
principalSource: disabled

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

Note

If you use the Event Consumer Service’s durable queues, see Event Consumer Dead-Letter Capture for the broker prerequisite (RabbitMQ 4.x / quorum queues), the dead-letter topology the service declares, the live-tunable numeric-limits policy, and the one-time rollout step for pre-existing classic durable queues.

Authentication and Permissions#

Configure identity (authentication) and authorization (permissions) for the Notifications Services — the Event Consumer Service and the Event Aggregation Service.

This section covers two orthogonal first-class controls and one consumer-only storage gate:

  • principalSource — selects how the caller’s identity (the principal) is derived from the incoming request, and whether a token is required at all. This is the single authentication gate for both services.

  • permissions.enabled — the authorization gate for the permission service path. Independent of authentication: when off, all permission checks short-circuit to “allowed”.

  • permissions.storage_permissions.enabled — a consumer-only authorization gate for the storage-service path. The Event Aggregation Service has no equivalent.

Note

Authentication and authorization are independent. You can derive a principal (principalSource set to a non-disabled mode) without enforcing any permissions, and you must derive a principal before you can enforce permissions. Enabling either permission gate while principalSource is disabled is an invalid configuration — there is no authenticated principal to authorize.

These keys live in your event-consumer-values.yaml and event-aggregation-values.yaml files (created in the deployment steps above). See Authentication for setting up the OIDC identity provider and Envoy auth extension.

Selecting the principal source (authentication)#

principalSource is a single first-class setting that selects how the principal (caller identity) is derived and whether a token is required.

Both services accept the same five modes. The setting maps to the OMNI_EVENTS_PRINCIPAL_SOURCE environment variable and defaults to jwtUnverified.

principalSource modes#

Mode

Token required?

Behavior

Required configuration

disabled

No

Authentication is off. No principal is derived and no token is required.

permissions.enabled must be false (and on the consumer, permissions.storage_permissions.enabled must be false). There is no principal to authorize.

jwtUnverified (default)

Yes

Decode the JWT to derive the principal without verifying its signature.

None beyond a token on the request. See the trust-boundary note below.

jwt

Yes

Validate the token as a JWT using public keys pulled from the identity provider.

openId.openIdConfigurationUri plus the accepted JWT audiences (via openId.clientRegistrations and/or openId.additionalJwtAudience).

opaque

Yes

Validate the token against the IdP /userinfo endpoint.

openId.openIdConfigurationUri.

userInfoService

Yes

Resolve the principal via an external user-info service over REST.

userInfoService.uri.

Every non-disabled mode requires a token on the request. Only disabled permits tokenless callers.

Warning

In jwtUnverified mode the service decodes the JWT to read the principal but does not verify the token’s signature. This is safe only when an upstream component (typically the Envoy auth extension described in Authentication) has already validated the token before it reaches the service. jwtUnverified trusts that upstream validation. Do not expose a service running in jwtUnverified mode directly to untrusted clients without that upstream validation in place — use jwt or opaque if the service must validate tokens itself.

disabled — no authentication#

# event-consumer-values.yaml or event-aggregation-values.yaml
principalSource: disabled
permissions:
  enabled: false   # required: no principal exists to authorize

jwtUnverified — decode without verifying (default)#

principalSource: jwtUnverified

Use this mode behind an Envoy auth extension that has already validated the token. See the trust-boundary warning above.

jwt — verify via identity-provider keys#

principalSource: jwt
openId:
  openIdConfigurationUri: "https://login.microsoftonline.com/{tenant-id}/v2.0/.well-known/openid-configuration"
  clientRegistrations:
    - name: "default"
      clientId: "{client-id}"
      scope: "openid profile email offline_access {client-id}/.default"
  # Use additionalJwtAudience when the IdP issues tokens whose `aud` is an
  # API/resource identifier rather than the client id:
  # additionalJwtAudience:
  #   - "{api-audience}"

opaque — validate via the IdP userinfo endpoint#

principalSource: opaque
openId:
  openIdConfigurationUri: "https://login.microsoftonline.com/{tenant-id}/v2.0/.well-known/openid-configuration"

userInfoService — resolve via an external user-info service#

principalSource: userInfoService
userInfoService:
  uri: "http://microsoft-graph-userinfo-service:8080"

Enabling permissions (authorization)#

Authorization is controlled by gates that are independent of authentication. When a gate is disabled, the corresponding permission checks short-circuit to “allowed” even if a principal was derived. When a gate is enabled, principalSource must not be disabled — authorization requires an authenticated principal.

The permission service gate: permissions.enabled#

permissions.enabled (env OMNI_EVENTS_PERMISSIONS_ENABLED, default false) gates enforcement against the permission service. It applies to both the Event Consumer Service and the Event Aggregation Service.

# event-consumer-values.yaml or event-aggregation-values.yaml
principalSource: jwtUnverified   # any non-disabled mode
permissions:
  enabled: true
  endpoint: "http://permission-service:3000"
  ttl_seconds: 600

The consumer-only storage gate: permissions.storage_permissions.enabled#

The Event Consumer Service has a second, independent authorization gate for storage events that flow through the storage-service path. permissions.storage_permissions.enabled (env OMNI_EVENTS_STORAGE_PERMISSIONS_ENABLED, default false) governs only the storage-service permission path and is independent of permissions.enabled.

# event-consumer-values.yaml (Event Consumer Service only)
principalSource: jwtUnverified   # any non-disabled mode
permissions:
  enabled: true
  storage_permissions:
    enabled: true
    endpoints:
      - "http://storage-service:8012"

Warning

The Event Aggregation Service has no storage gate. permissions.storage_permissions exists only in the Event Consumer Service chart; do not add it to event-aggregation-values.yaml.

How the two gates split by code path#

On the Event Consumer Service, storage events are authorized differently depending on the queue type the client consumes from:

  • Durable-queue storage events are authorized via the permission service — gated by permissions.enabled.

  • Non-durable-queue storage events are authorized via the storage service — gated by permissions.storage_permissions.enabled.

The two gates are independent. To enforce permissions on both durable and non-durable storage-event consumption, enable both. Enabling only permissions.enabled leaves non-durable storage-event checks short-circuited to “allowed”; enabling only permissions.storage_permissions.enabled leaves durable-queue (and other permission-service) checks short-circuited to “allowed”.

Note

The storage-service endpoints are configured via permissions.storage_permissions.endpoints (an array) and are consulted only when permissions.storage_permissions.enabled is true. They are rendered into the service’s permissions YAML file at deploy time.

Putting it together#

A common production configuration for the Event Consumer Service, behind a validated Envoy auth extension, enforcing permissions on both durable and non-durable storage events:

# event-consumer-values.yaml
principalSource: jwtUnverified
permissions:
  enabled: true
  endpoint: "http://permission-service:3000"
  storage_permissions:
    enabled: true
    endpoints:
      - "http://storage-service:8012"

The equivalent Event Aggregation Service configuration has no storage gate:

# event-aggregation-values.yaml
principalSource: jwtUnverified
permissions:
  enabled: true
  endpoint: "http://permission-service:3000"

event-consumer-values.yaml

event-aggregation-values.yaml

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#

  1. In the AWS Console, open the SQS Dashboard and click Create queue.

  2. Set Type → Standard, Name, and Message retention period (e.g. several hours).

  3. 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.

  4. Create the queue and note its URL.

Setting up an Event Subscription (S3 → SQS)#

  1. In the AWS Console open S3, select the bucket, then Properties → Event notifications.

  2. 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.

  3. 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)#

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

  2. 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.

  3. 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 SNS topic#

Create the topic in each region where you have S3 buckets.

  1. In the AWS Console open SNS → Topics → Create Topic. Choose Standard, set Name.

  2. Access Policy → Advanced. Add one statement per S3 bucket in this region:

    {
  "Sid": "<statement-id>",
  "Effect": "Allow",
  "Principal": { "Service": "s3.amazonaws.com" },
  "Action": "SNS:Publish",
  "Resource": "arn:aws:sns:<aws-region>:<aws-account-id>:<sns-topic-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>, <sns-topic-name>, <s3-bucket-name>.

  3. Create the topic and note its ARN.

  4. Open the Subscriptions tab → Create Subscription. Set Topic ARN (from step 3), Protocol → Amazon SQS, Endpoint → ARN of the SQS queue from the previous subsection. Enable raw message delivery must be enabled. Create the subscription.

Setting up an Event Subscription (S3 → SNS)#

Repeat per S3 bucket:

  1. S3 → select bucket → Properties → Event notifications → Create event notification.

  2. 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.

  3. Save changes.

Validation (AWS)#

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

  1. SQS Dashboard → open your queue → Send and receive messages.

  2. Receive Messages → Edit poll settings → set Polling duration to 1 second → Poll for messages.

  3. 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)

  1. 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.

  2. 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.

  3. 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#

  1. Log into the Azure Portal.

  2. 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.

  3. In the Service Bus Namespace view click + Queue.

  4. Set Name and Message time to live (e.g. several hours instead of the default 14 days). Review and create the queue.

  5. In the Storage Accounts browser, open the Storage Account for which you want the event subscription.

  6. In the left panel, open Events and click + Event Subscription.

  7. 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.

  8. 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

This walkthrough is tokenless: the curl commands below send no Authorization header. They succeed only if the Event Consumer and Event Aggregation services were deployed with principalSource: disabled (see the configuration steps above). Under the default jwtUnverified mode the stream request is rejected with an authentication error. To test with authentication enabled, see the Authentication and Permissions section above.

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/...

Subscribe to Events (Terminal 1)#

First, set up a subscription to watch all storage events at the root level. This uses a starts_with_greedy filter with an empty resource_id to match all resources.

# Define filters for all storage create and delete events
FILTERS='[
  {"event_type":"omni.storage.created","filters":[{"filter_type":"starts_with_greedy","resource_id":""}]},
  {"event_type":"omni.storage.deleted","filters":[{"filter_type":"starts_with_greedy","resource_id":""}]}
]'

# Start listening for events (this will stream events as they occur)
curl -N -G "http://event-consumer-service.storage-apis.svc.cluster.local:8000/api/v1beta/events/stream" \
  -H "Accept: text/event-stream" \
  --data-urlencode "filter_groups=$FILTERS"

Note

The -N flag disables buffering so events appear immediately. This command will run continuously, streaming events as they occur. Press Ctrl+C to stop.

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

On this page