Kubernetes Controller

The JetStream controllers allow you to manage NATS JetStream Streams and Consumers via K8S CRDs. You can find more info on how to deploy and usage here. Below you can find an example of how to create a stream and a couple of consumers:

  1. ---
  2. apiVersion: jetstream.nats.io/v1beta1
  3. kind: Stream
  4. metadata:
  5.   name: mystream
  6. spec:
  7.   name: mystream
  8.   subjects: ["orders.*"]
  9.   storage: memory
  10.   maxAge: 1h
  11. ---
  12. apiVersion: jetstream.nats.io/v1beta1
  13. kind: Consumer
  14. metadata:
  15.   name: my-push-consumer
  16. spec:
  17.   streamName: mystream
  18.   durableName: my-push-consumer
  19.   deliverSubject: my-push-consumer.orders
  20.   deliverPolicy: last
  21.   ackPolicy: none
  22.   replayPolicy: instant
  23. ---
  24. apiVersion: jetstream.nats.io/v1beta1
  25. kind: Consumer
  26. metadata:
  27.   name: my-pull-consumer
  28. spec:
  29.   streamName: mystream
  30.   durableName: my-pull-consumer
  31.   deliverPolicy: all
  32.   filterSubject: orders.received
  33.   maxDeliver: 20
  34.   ackPolicy: explicit

Once the CRDs are installed you can use kubectl to manage the streams and consumers as follows:

  1. $ kubectl get streams
  2. NAME       STATE     STREAM NAME   SUBJECTS
  3. mystream   Created   mystream      [orders.*]
  5. $ kubectl get consumers
  6. NAME               STATE     STREAM     CONSUMER           ACK POLICY
  7. my-pull-consumer   Created   mystream   my-pull-consumer   explicit
  8. my-push-consumer   Created   mystream   my-push-consumer   none
  10. # If you end up in an Errored state, run kubectl describe for more info.
  11. #     kubectl describe streams mystream
  12. #     kubectl describe consumers my-pull-consumer