Proposal: Add AI/ML-based network anomaly detection stage to flowlogs-pipeline

[vatankh]

Component(s)

flowlogs-pipeline / enrichment stages

Problem

NetObserv provides excellent collection and aggregation of Kubernetes network flow logs,
but the current pipeline stops at aggregation and visualization. There is no native stage
that detects operational anomalies such as sudden throughput spikes, new/unexpected
service-to-service edges, or latency regressions.

Proposed solution

Introduce an optional "anomaly detection" or "smart enrichment" stage that:

• Consumes aggregated flow records (from Loki, Kafka, or direct pipeline input).

• Applies simple streaming statistics (EWMA, z-score, moving average) or lightweight ML models
  to compute an anomaly_score per (src, dst, namespace, protocol).

• Emits enriched flow records with fields like:

  {
    "anomaly_score": 0.87,
    "anomaly_type": "throughput_spike",
    "baseline_window": "5m"
  }

• Allows sensitivity tuning through configuration (e.g., YAML or environment variables).

• Outputs can be visualized in Grafana or fed into alerting systems.

Benefits

• Extends NetObserv’s analytics capabilities beyond raw metrics.

• Provides proactive detection of performance regressions or misconfigurations.

• Stays within observability scope (not IDS/security).

[jotak]

Hi @vatankh , thanks for opening this issue. I agree that there is a good opportunity here. Just to clarify: is it something that you propose to implement yourself?

A remark on this:

Consumes aggregated flow records (from Loki, Kafka, or direct pipeline input).

Consuming exported records (from Loki or Kafka) is probably out of the scope of flowlogs-pipeline, which today is fully on the ingestion side. But the idea would still makes sense when we consider processing data from the configured pipeline input.

If we think it's better to consume from Loki/Kafka, we could rather consider creating a new component dedicated to that task.

Also, flowlogs-pipeline is generally partitioned between several instances, thus it has access to just a fraction of the whole data. Which I believe makes the Loki/Kafka sources more relevant for AI/learning.

[vatankh]

Thanks a lot for the detailed feedback @jotak — that makes perfect sense.

Yes, I’m proposing to implement this myself as a contribution.

Based on your clarification, I’ll focus the initial prototype on processing data directly from the existing pipeline input, without pulling from external stores (Loki/Kafka).
That way it stays aligned with the current architecture.

The goal for the first version would be:

• Add a new enrichment stage that computes simple statistical anomaly scores (EWMA, z-score, etc.) per flow record within a partition.
• Emit additional fields like anomaly_score and anomaly_type, which can then be visualized or exported as usual.

Does that approach sound good to you?

[jotak]

Yes, that sounds good. Let me know if you need any help/guidance.

[vatankh]

@jotak Thanks again for the guidance!

I wanted to give an update: I have now implemented the initial version of the anomaly-detection transform that we discussed. It introduces a new type: anomaly stage that performs streaming anomaly scoring (z-score or EWMA) on the existing pipeline input, without consuming data from Loki/Kafka.

The implementation includes:

• configurable algorithm, window size, baseline window, sensitivity, keyFields, etc.
• per-key rolling baseline and anomaly scoring
• added fields (anomaly_score, anomaly_type, baseline_window) on each flow record
• test coverage + example pipeline configuration + API documentation

The pull request is now open here:

#1143

Feedback is very welcome — I’m happy to make adjustments or improvements based on your review or any architectural preferences.

Thanks again for the help and for maintaining this project!

[jotak]

Thanks @vatankh, I'm looking into it