Use Message Streaming with other products on CloudFerro Cloud
After processing a dataset, a worker needs to tell the next application where to find the result. This example publishes a small event containing an object location, then reads it with the reporting consumer. The same publication can run from a VM, container, or workflow task; the final checks show how to exercise a Functions connection.
Application patterns
These are examples you can implement with application code and the products’ connection settings:
OpenStack virtual machines: processing programs on VMs publish progress events through a NATS client. A dashboard subscribes to updates, while a reporting consumer reads retained events to calculate job durations.
Managed Kubernetes: run several services as containers, with separate consumers for validation, indexing, and reporting. Each service can read the retained events it needs and track its own progress.
Private Object Storage and Functions: an uploader stores a dataset in a bucket, then publishes its object key to NATS. A NATS trigger invokes a function whose code retrieves the object using S3 credentials and checks its metadata.
Workflows: an Airflow task publishes an event after a scheduled processing stage completes. Downstream applications consume the event when ready, and a retained history records which datasets completed that stage.
VM, Kubernetes, and Workflows applications use a NATS client and account credentials. For Functions, configure a NATS trigger or output in My Portal. Message Streaming can also be used independently of Functions and Message Broker.
Start with the CLI example below before putting the publication into an application or connecting a function.
Prerequisites
No. 1 The example stream and a working connection
Complete Work with streams and consumers on CloudFerro Cloud. Have JOB_STATUS, jobs.status, and the reporting consumer available, with NATS_URL and NATS_CREDS set in your terminal. The reporting consumer should have read the earlier exercise’s messages.
No. 2 An object location when retrieving a real dataset
The messaging test can use the illustrative bucket and key below. To retrieve an actual object, upload it first and substitute its real location. The receiving application needs the S3 endpoint and its own access to that bucket. A NATS credentials file does not provide S3 access.
No. 3 An existing function for the optional Functions checks
Your account needs access to Functions and Message Streaming in the organization containing the resources. Prepare a function using Create and deploy a function on CloudFerro Cloud, and configure its messaging connection using Configure function triggers and outputs on CloudFerro Cloud. Have a running function in the same region as the NATS account. The function’s code must understand the events you send.
Publish the location of a result
Suppose job-002 has finished and its result is stored under jobs/job-002/result.json in example-results. Publish the location rather than the file contents:
nats --tlsfirst pub jobs.status \
'{"job_id":"job-002","status":"finished","bucket":"example-results","key":"jobs/job-002/result.json"}' \
--jetstream
Check that the response identifies JOB_STATUS and a sequence number. This confirms storage of the event. The command does not upload the object or verify that the bucket and key exist.
Read the event as the reporting application
Fetch the new event with the reporting consumer:
nats --tlsfirst consumer next JOB_STATUS reporting \
--count=1 --ack --raw
The payload should contain the job identifier and object location:
{
"job_id": "job-002",
"status": "finished",
"bucket": "example-results",
"key": "jobs/job-002/result.json"
}
The CLI displays compact JSON; the example above is expanded for readability. If an older event appears, the consumer had earlier messages waiting. Read its remaining messages before checking for this job identifier.
An application would parse bucket and key, retrieve the object with its S3 client, process it, and acknowledge the NATS delivery after processing succeeds. The CLI exercise acknowledges on receipt; it checks message delivery without retrieving the object.
Use the publication in a worker or workflow
Place the publication after the processing and upload steps succeed. A VM or container can run the same CLI command, while a Workflows task can execute it or use a NATS SDK.
Set the connection values in that execution environment and make the credentials file readable by its process. Exporting variables in your desktop terminal does not configure a remote worker. Use the job’s actual identifier and result location in its event.
Keep failed processing or upload steps from publishing a finished event. Check the publication’s exit status and retain the job identifier in the worker’s logs so that a delivery can be traced back to the processing task.
Check the regions
Functions and the messaging resource must be in the same region. Region selection is stored separately for Functions, Message Broker, and Message Streaming, so changing it in one product does not change it in the others. Check both resources before configuring the trigger or output.
Resources are not replicated between regions automatically. Use the connection values for the region where the resources actually run.
Test a NATS input to a function
Open the function’s Triggers and output tab. Configure a NATS input using the account connection and jobs.status subject.
Run Test configuration. Resolve region or credential errors before sending an event.
Publish the result-location event above with a distinct job identifier, such as function-001.
Check the function’s logs for an invocation at the publication time. If its handler logs the job identifier, look for function-001 and check the processing outcome.
A successful connection test verifies the connection settings; the event and invocation check exercise the handler. If the integration uses a named consumer, give it its own consumer and avoid fetching or acknowledging that consumer’s messages with the CLI.
If no invocation appears, first confirm that the event reached the stream:
nats --tlsfirst stream get JOB_STATUS --last-for=jobs.status
Compare the latest payload’s job_id with the event you sent. Reading a stored message this way does not advance the function’s consumer.
Watch a function output arrive
For an output test, configure the function’s NATS output with an unused subject such as jobs.function.result. In a terminal, start a live subscriber before invoking the function:
nats --tlsfirst sub jobs.function.result --count=1
Invoke the function with its test input. The subscriber should show the message published through the output connection and exit after one delivery. Inspect that payload against the function’s output configuration; it is not necessarily identical to its input event. This live check does not create retained storage for the output subject.
What to do next
Use Test a function and manage its endpoints on CloudFerro Cloud for function test requests and Monitor and troubleshoot Functions on CloudFerro Cloud for logs. Continue with Monitor and remove a NATS account on CloudFerro Cloud to inspect stream storage and each reader’s progress.