This PR implements CFP-37472

The goal is to allow user to aggregate flows on desired properties to reduce the overall volume of logs produced to in turn reduce storage costs.
This PR introduces flow aggregation capabilities to Hubble's dynamic and static exporters and enabling collection and export of aggregated flows.

The PR may seem large but I think keeping this together makes sense. It consists of the following parts:
API changes (5): 041e12d
Aggregation Logic (14): 025bfa2
Charts & Docs (7): e7b837a
Exporter Fix (rep): d393cec

Overview

Supports multiple dynamic flowlog configs.

• pkg/hubble/exporter/aggregator.go - AggregationLogic. Thread-safe aggregation with field-based grouping and direction-aware flow counting
• pkg/hubble/exporter/config.go - Configuration parsing with new 'Duration' wrapper type for YAML/JSON unmarshaling. This is needed for Dynamic Exporters hot-reload 'Duration' parsing in order to be able to use '10s/10m' duration formats.
• pkg/hubble/parser/fieldaggregate/ - Field Aggregate is a wrapper that extends FieldMask functionality.
• api/v1/aggregate/aggregate.proto - New aggregate message definitions for flow counts. This will only be used when Agrgegation feature is enabled and will be logged as part of flow.Aggregate.
  • Adding aggregation fields directly to the Flow api.

message Aggregate {
  uint32 ingress_flow_count = 1;
  uint32 egress_flow_count = 2;
  uint32 unknown_direction_flow_count = 3;
}

• pkg/hubble/exporter/context_cancellation_test.go - Testing of graceful shutdown and goroutine management to ensure no go routine leaks.

Config update

two new fields under hubble.export.static && hubble.export.dynamic.

• fieldAggregate: ["source.namespace", "destination.service_name"]
• aggregationInterval: "1m"
  For Dynamic exporter a change in either will trigger hot reload.
  If aggregationInterval is set to 0s - aggregation is disabled
  If both FieldAggregate and FieldMask is enabled. FieldMask is ignored. (More in FieldMask & FieldAggregate analysis below)

FieldMask & FieldAggregate analysis:

FieldAggregate config:

fieldAggregate={source.namespace,source.pod_name,destination.namespace,verdict}"
Aggregated Flow structure:

"flow":{
   "verdict":"FORWARDED",
   "source":{
      "namespace":"default",
      "pod_name":"kapinger-bad-666498f4bc-w262k"},
   "destination":{
      "namespace":"default"
   },
   "aggregate":{
      "ingress_flow_count":78,
      "egress_flow_count":345,
      "unknown_direction_flow_count":27
   }
}

ConfigMap: (FieldMask config is ignored)

  flowlogs.yaml: |
    flowLogs:
    - aggregationInterval: 30s
      fieldAggregate:
      - source.namespace
      - source.pod_name
      - destination.namespace
      - verdict
     fieldMask:
      - time
      - source.namespace
      - source.pod_name
      - destination.namespace
      - destination.pod_name
      - l4
      - IP
      - node_name
      - is_reply
      - verdict
      - drop_reason_desc
      filePath: /var/run/cilium/hubble/events-agg.log
      includeFilters:
      - source_pod:
        - default/
      - destination_pod:
        - default/
      name: system

Testing Steps:

Manual Verrification:
BYO CNI Cilium Cluster

• helm upgrade cilium cilium/cilium --version 1.19-dev
• Build local images and update cluster
• Deploy a workload that produces alot of pings to itself
• Test aggreagtion via various combinations or below.

   --set "hubble.export.dynamic.config.content[1].aggregationInterval=25s" \
   --set "hubble.export.dynamic.config.content[1].fieldMask={time,source.namespace,source.pod_name,destination.namespace,destination.pod_name,l4,IP,node_name,is_reply,verdict,drop_reason_desc}" \
   --set "hubble.export.dynamic.config.content[1].fieldAggregate={source.namespace,source.pod_name,destination.namespace}" \
   --set "hubble.export.dynamic.config.content[1].name=agg2" \
   --set "hubble.export.dynamic.config.content[1].filePath=/var/run/cilium/hubble/events-agg2.log" \
   --set "hubble.export.dynamic.config.content[1].includeFilters[0].source_pod[0]=default/" \
   --set "hubble.export.dynamic.config.content[1].includeFilters[1].destination_pod[0]=default/"

• Test with multiple configs
• Test aggregation with static exporter
• Verification of cilium-flowlog-config ConfigMap
• Verification of logs in multiple files via node-shell
  • var/run/cilium/hubble
  • check that file names match CM
  • check for aggregation counts
  • check for config reloads
  • check setting to 0s disables aggregation

Support for Configurable Hubble FlowLog Aggregation (CFP-37472)