Support for alternative message queues

[stone-z]

Overview

The synchronizer currently uses an Apache Pulsar message queue to synchronize state between storage and individual instances of kubescape.

Pulsar can support a very large message volume, has nice features for transforming messages in transit, as well as a number of supported input and output connectors.

Problem

Pulsar seems like a really nice tool for this use case, but is not as widely used as some alternatives.

Specifically, it might make sense to support Kafka, since it has very widespread adoption, or NATS, because it is also a CNCF project and has a smaller resource footprint.

Other "lighter" message queues may also be interesting for users who want the basic functionality of the synchronizer, but without the footprint of Pulsar (or Kafka, for that matter).

Solution

This issue is mostly to inquire whether the project is open to contributions for alternative backend implementations, and to get a general understanding about what the maintainers would want to see from such contributions. Would it be in line with the project's long-term goals? Are there requirements that a potential backend would need to fulfill?

Alternatives

Only support Pulsar (status quo).
Users could fork or re-implement the synchronizer and add their own backend support.
Users could connect Pulsar to a desired alternative using Pulsar sources/sinks.

Additional context

N/A

[matthyx]

@stone-z are you running the synchronizer server? I'm interested in your use-case...

Appart from that, I agree we should abstract the queue.

[yugal07]

Hi @matthyx, happy to take a pass once @stone-z's use case is clearer (don’t want to cut across that thread).

Reading through the code, the abstraction for a pluggable backend is largely already in place (messaging.MessageProducer / MessageReader), with Pulsar as the only implementation today. The main remaining coupling:

• config.Backend.PulsarConfig is Pulsar-specific rather than a discriminated backend.
• cmd/server/main.go wires Pulsar directly (no factory).
• Some helpers leak Pulsar types/semantics:
  • utils.PulsarMessageIDtoString
  • synchronizer_pulsar_producer_* metrics
  • self-loop filter via message Key (can move to properties for portability)
• EnableChunking: true suggests reconciliation messages can exceed typical broker limits — any alternative (NATS/Kafka) will need chunking or splitting.
• reconciliationMessageMutex is broker-agnostic but currently lives inside the Pulsar implementation.

Proposed direction:

• Introduce Backend.MessageQueue as a discriminated config (type: pulsar | nats | kafka) with a small factory returning (MessageProducer, MessageReader).
• Move message-id handling, self-loop filtering (via a property), reconciliation mutex, and metrics into the messaging package as backend-agnostic helpers.
• Land the refactor with Pulsar only, then add NATS JetStream as the first alternative (Kafka later, given differing consumer semantics).

Open questions:

• Delivery semantics: at-least-once + idempotent consumers, or reliance on Pulsar ordering/exclusivity?
• Ordering scope: per (account, cluster) or per topic?
• Max message size: do we need app-level chunking?
• Multi-tenancy: is Pulsar tenant/namespace operator-facing or internal?

If this direction makes sense, happy to open a short design doc PR before any code changes.

[matthyx]

Thanks for jumping in @yugal07 ! Let's hear first from @stone-z 👍

[stone-z]

Thanks to you both for the quick replies!

First a disclaimer: this is still theoretical; I'm not up and running with Pulsar and haven't worked on it at a technical level yet. It's possible this is a Bad Idea ™️ and I just haven't figured that out yet. That said...

My interest in the synchronizer is that I want to [ab]use it and the underlying message queue for shipping security data from multiple tools around in different cluster topologies. My team manages several CNCF security projects for clients. Some common unmet needs we see are:

• a client wants a fleet-wide overview of security data, but individual tools typically store data only in the local cluster
• a client wants to send security data to some external place, perhaps transforming it (redacting, down-sampling, etc.) first
• a client has [semi-] air-gapped clusters included in their landscape that they'd still like to include in reporting

In order to meet those needs, my high-level design calls for running a message queue in each cluster to receive and buffer security data from in-cluster tools, route the data to a local data store, and additionally transform and route it to any configured external locations. Specifically, I would use the queue to ingest Falco events and Kyverno PolicyReports locally in each workload cluster and configure a management cluster queue/data store plus any customer-provided external data stores as consumers of the queue.

I could build some adapters and use Pulsar to do that, no doubt about it. However, I anticipate some resistance to Pulsar because of its resource footprint. Running a Pulsar-per-cluster is fairly heavy, but if I could run a lightweight queue in each monitored cluster which exists only to buffer and ship data to a beefier queue, that would be more palatable. My hope is to run and support only one message queue that could be used for both kubescape and our other tools.

The ✨ Dream ✨ is to converge all of the CNCF security/policy/compliance tools to a standard report format and common storage location (perhaps kubescape storage or Kyverno reports-server), and use a common mechanism to work with them in a fleet. This is a similar motivation as in our past discussion about PolicyReport support. Supporting other queues would be a step in that general direction.

WDYT?

[matthyx]

In general I'm not against it, I like the long term goal, and synchronizer is supposed to be generic enough to support it.
How do you see the next steps?

[stone-z]

On my side, we we're still rolling out kubescape (just local installations without pulsar) to the environments that would benefit most from this. We expect to have Kafka available to us internally on the order of ~months. There are some more roadmap items we need to clear between now and then, but once Kafka is running we would look to start sending kubescape data there.

If it's simply a question of implementation, if someone else has time and interest, go for it! but we are happy to contribute it once we get there.

Otherwise, let me know how we can help. I agree with @yugal07 that the codebase seems pretty easily extensible. I am happy to review a design doc or join a sync to offer any other info, use cases, etc.

[harshitg927]

Hello @stone-z , @matthyx and @yugal07. I had a discussion regarding this issue with Yugal on slack, as he is currently a little occupied with LFX right now, I will be taking a look into this issue and soon be opening a design document PR for this.
I hope this works, Thanks!

[harshitg927]

Hi @matthyx @stone-z @yugal07 — I've opened a design doc proposal for pluggable message queue support with Kafka as the first alternative backend:

Design doc PR: kubescape/designs-and-proposals#7

The proposal covers:

• Current architecture and what is already abstracted (MessageProducer / MessageReader)
• Discriminated config schema with Pulsar backward compatibility
• Split-topic Kafka design (synchronizer.out / synchronizer.in)
• Fan-out consumer group semantics (critical for multi-pod routing)
• Phased implementation plan (refactor → Kafka → event-ingester → Helm)
• Cross-repo coordination with event-ingester-service

Happy to adjust based on feedback. Let me know if a sync would be helpful before implementation starts.