README.STREAMDAL.md

Pika RabbitMQ Client (instrumented with Streamdal)

This library has been instrumented with Streamdal's Python SDK.

Getting Started

The following environment variables must be set before launching a producer or consumer:

1. STREAMDAL_ADDRESS
   • Address for the streamdal server (Example: localhost:8082)
2. STREAMDAL_TOKEN
   • Authentication token used by the server (Example: 1234)
3. STREAMDAL_SERVICE_NAME
   • How this application/service will be identified in Streamdal Console (Example: billing-svc)

By default, the library will not have Streamdal instrumentation enabled; to enable it, you will need to set enable_streamdal=True in the pika connection params:

pika.ConnectionParameters(enable_streamdal=True)

_{For more in-depth explanation of the changes and available settings, see What's changed?.}

Example

A fully working example is provided in examples/streamdal.

To run the example:

1. Change directory to examples/streamdal
2. Start a local rabbitmq instance: docker-compose up -d rabbitmq
3. Install & start Streamdal: curl -sSL https://sh.streamdal.com | sh
4. Open a browser to verify you can see the streamdal UI at: http://localhost:8080
   • It should look like this:
5. Launch a consumer:

   STREAMDAL_ADDRESS=localhost:8082 \
   STREAMDAL_TOKEN=1234 \
   STREAMDAL_SERVICE_NAME=demo \
   python consumer.py
   

6. In another terminal, launch a producer:

   STREAMDAL_ADDRESS=localhost:8082 \
   STREAMDAL_TOKEN=1234 \
   STREAMDAL_SERVICE_NAME=demo \
   python producer.py
   

7. Open the Streamdal Console in a browser https://localhost:8080
   • It should look like this:
8. Create a pipeline that detects and masks PII fields & attach it to the consumer
   • 
9. Produce a message in producer terminal: testKey:{"email":"[email protected]"}
10. You should see a masked message in the consumer terminal: {"email":"fo*********"}
    • Tip: If you detach the pipeline from the consumer and paste the same message again, you will see the original, unmasked message.

Passing "runtime" settings to the shim

By default, the shim will set the component_name to rabbitmq and the operation_name to the name of the routing key you are producing to, or the string f"{queue}_{binding_key}" being consumed from

Also, by default, if the shim runs into any errors executing .process(), it will swallow the errors and return the original value.

When producing, you can set streamdal_cfg in the basic_publish():

cfg_produce = StreamdalRuntimeConfig(
   audience=Audience(
      component_name="rabbitmq",
      operation_name="produce_msg",
      operation_type=OPERATION_TYPE_PRODUCER
   )
)

channel.basic_publish(
   exchange='',
   routing_key='test',
   body='Hello World!',
   streamdal_cfg=cfg_produce
)

When consuming, you can set streamdal_cfg in the consume() call:

cfg_consume = StreamdalRuntimeConfig(
   audience=Audience(
      component_name="rabbitmq",
      operation_name="consume_msg",
      operation_type=OPERATION_TYPE_CONSUMER
   )
)

for method_frame, properties, body in channel.consume(queue='test', streamdal_cfg=cfg_consume):
    #...

What's changed?

The goal of any shim is to make minimally invasive changes so that the original library remains backwards-compatible and does not present any "surprises" at runtime.

1. consume() now accepts an optional streamdal_cfg parameter
2. basic_publish() now accepts an optional streamdal_cfg parameter
3. ConnectionParameters now accepts an optional enable_streamdal boolean parameter