docs/STREAMING_IMPLEMENTATION.md
This implementation addresses the API-server load issue described in A1 of issues.md and Phase 5 of docs/optimization-plan.md. The solution introduces resource streaming so the OPA evaluation never holds the whole cluster as its input: resources are partitioned by scope and the processor evaluates the resident (cluster-scoped) batch plus one namespace batch at a time, while the collection phase replaces the previous O(L × N) per-GVR-per-namespace LIST calls with a single LIST per GVR (O(L)).
Note what streaming does not do: it does not reduce total retained memory. The collector still pulls every GVR and holds the full cluster until the first batch is sent, and the processor retains all resources in AllResources for downstream stages (exceptions, printers, image scanning). Streaming bounds the evaluation input and the number of API-server calls, not the process high-water mark.
Kubescape previously loaded the entire cluster state into AllResources before evaluating anything. On clusters larger than ~2500 resources, this reached 2-4 GB of memory usage. The codebase contained an explicit admission of this workaround:
// isLargeCluster returns true if the cluster size is larger than the largeClusterSize
// This code is a workaround for large clusters. The final solution will be to scan resources individually
The implementation leverages the existing ResourceBatch architecture that was already present in the codebase but was being used only for scope-based evaluation, not memory management. The solution adds a streaming interface that:
Partitions resources by scope: Cluster-scoped resources (Nodes, ClusterRoles, etc.) are kept resident in memory throughout the scan, while namespace-scoped resources are processed in batches.
Streams resources incrementally: Instead of loading all resources at once, resources are streamed in batches via channels.
Bounded evaluation input: Each namespace batch is evaluated one at a time against the resident batch, so the evaluation input never contains the whole cluster.
Maintains result parity: The streaming implementation produces identical results to the non-streaming approach.
core/pkg/resourcehandler/interface.go)Added a new method to the IResourceHandler interface:
StreamResourcesBatches(ctx context.Context, sessionObj *cautils.OPASessionObj, scanInfo *cautils.ScanInfo) (<-chan *cautils.ResourceBatch, <-chan error, error)
This method returns channels for receiving resource batches and errors, enabling the caller to process resources incrementally.
core/pkg/resourcehandler/k8sresources.go)Implemented StreamResourcesBatches for Kubernetes resources with a two-phase approach:
The resident batch includes:
Namespace batches contain only the resources belonging to a specific namespace, so the evaluation input never holds more than the resident batch plus one namespace.
core/pkg/resourcehandler/filesloader.go)For file-based resources (typically smaller), the implementation loads all resources and returns them as a single batch for simplicity, since file-based scans don't typically have memory issues.
core/pkg/opaprocessor/processorhandler.go)Added ProcessWithStreaming method that:
The method leverages the existing evaluationScope and matchedObjects logic, ensuring that related-object resolution works correctly across batches since cluster-scoped resources remain resident.
core/core/scan.go, cmd/scan/scan.go)Added CLI flag --enable-streaming to manually enable streaming, and auto-detection for large clusters:
scanCmd.PersistentFlags().BoolVar(&scanInfo.EnableStreaming, "enable-streaming", false, "Enable resource streaming for large clusters. Resources are collected in a single pass per type and evaluated one namespace at a time. Automatically enabled for clusters with >2500 resources.")
The scan logic automatically enables streaming for clusters with >2500 resources (configurable via LARGE_CLUSTER_SIZE environment variable).
core/cautils/scaninfo.go)Added EnableStreaming field to ScanInfo struct to control streaming behavior.
Leverages existing architecture: The ResourceBatch and PartitionResources logic already existed, reducing the risk of introducing bugs.
Maintains correctness: By keeping cluster-scoped resources resident, related-object resolution continues to work correctly since rules can access cluster-scoped resources from any namespace batch.
Deterministic ordering: Namespace batches are processed in sorted order, ensuring reproducible results.
Graceful degradation: If streaming fails, the system can fall back to the traditional approach.
The implementation reduces peak evaluation memory by:
The collection peak is not reduced: all resources are pulled and partitioned before the first batch is sent, and the processor retains resources in AllResources for downstream stages. The real win over the previous approach is the API-server load — one LIST per GVR instead of one per GVR per namespace — and a bounded evaluation input.
The existing matchedObjects function already handles cross-scope resolution correctly:
func (scope evaluationScope) matchedObjects(rule *reporthandling.PolicyRule) []workloadinterface.IMetadata {
var objects []workloadinterface.IMetadata
if scope.batch != nil {
objects = getKubernetesObjects(scope.batch.K8SResources, scope.batch.AllResources, rule.Match)
if len(objects) == 0 {
return nil
}
}
objects = append(objects, getKubernetesObjects(scope.resident.K8SResources, scope.resident.AllResources, rule.Match)...)
objects = append(objects, getKubernetesObjectsFromExternalResources(scope.resident.ExternalResources, scope.resident.AllResources, rule.DynamicMatch)...)
return objects
}
Since scope.resident contains all cluster-scoped resources, rules can access them regardless of which namespace batch is being processed.
Added comprehensive parity tests in core/pkg/opaprocessor/processorhandler_streaming_test.go:
The tests confirm that:
For a cluster with 2500 namespaces and 100 GVRs, this is ~250,000 LIST calls down to 100.
Minimal CPU overhead from:
The streaming approach may be slightly slower due to the overhead of managing batches, but the API-server LIST savings are significant for large clusters.
kubescape scan --enable-streaming
Streaming is automatically enabled for clusters with >2500 resources:
export LARGE_CLUSTER_SIZE=2500 # Default threshold
kubescape scan # Will auto-enable streaming for large clusters
--enable-streaming=falseLARGE_CLUSTER_SIZEThis implementation addresses the API-server load issue described in A1 by replacing the O(L × N) collection loop with a single pass per GVR and by streaming batches to the OPA processor so evaluation never sees the whole cluster as its input. The solution maintains correctness while keeping the evaluation input bounded, making Kubescape more suitable for scanning enterprise-scale Kubernetes environments. It is not a total-process memory reduction: bounding the collection peak itself is left to future paged-LIST collection.