bwmap: Reconcile cilium_throttle with StateDB reconciler

Store the per-endpoint EDT settings in Table[bwmap.Edt] and
reconcile it to the cilium_throttle BPF map with the StateDB
reconciler.

This is the first BPF map being converted over to this style
and aims to serve as an example. The overall point is to use
a more general mechanisms for recilience and status rather than
ones specific to pkg/bpf.

This improves resilience (failed operations are retried, and
the map re-synced periodically, similar to WithCache()).
The reconciliation status of the entries can inspected with
"cilium-dbg statedb bandwidth-edts" and the failed reconciliation
is reported as part of module health ("cilium-dbg status --all-health")

The replacements to existing functionality in pkg/bpf.Map map
as follows:

- WithEvents => cilium-dbg statedb bandwidth-ets --watch=100ms
- WithCache => Table[bwmap.Edt] & reconciler
- WithPressureMetric => RegisterTableMetricsJob

Signed-off-by: Jussi Maki <[email protected]>

Documentation/spelling_wordlist.txt

@@ -295,6 +295,7 @@ eBGP
 eBPF
 eXpress
 ebpf
+edts
 effectful
 egressing
 elfutils

cilium-dbg/cmd/statedb.go

@@ -23,6 +23,7 @@ import (
  "github.com/cilium/cilium/pkg/datapath/tables"
  "github.com/cilium/cilium/pkg/healthv2"
  "github.com/cilium/cilium/pkg/healthv2/types"
+ "github.com/cilium/cilium/pkg/maps/bwmap"
 )
 
 var StatedbCmd = &cobra.Command{
@@ -137,6 +138,7 @@ func init() {
  statedbTableCommand[*tables.Sysctl](tables.SysctlTableName),
  statedbTableCommand[types.Status](healthv2.HealthTableName),
  statedbTableCommand[*tables.IPSetEntry](tables.IPSetsTableName),
+ statedbTableCommand[bwmap.Edt](bwmap.EdtTableName),
  )
  RootCmd.AddCommand(StatedbCmd)
 }

pkg/datapath/fake/types/bandwidth.go

@@ -9,6 +9,12 @@ var _ types.BandwidthManager = (*BandwidthManager)(nil)
 
 type BandwidthManager struct{}
 
+func (fbm *BandwidthManager) DeleteBandwidthLimit(endpointID uint16) {
+}
+
+func (fbm *BandwidthManager) UpdateBandwidthLimit(endpointID uint16, bytesPerSecond uint64) {
+}
+
 func (fbm *BandwidthManager) BBREnabled() bool {
  return false
 }

pkg/datapath/linux/bandwidth/bandwidth.go

@@ -18,6 +18,7 @@ import (
 
  "github.com/cilium/cilium/pkg/datapath/linux/config/defines"
  "github.com/cilium/cilium/pkg/datapath/linux/probes"
+ "github.com/cilium/cilium/pkg/datapath/types"
  "github.com/cilium/cilium/pkg/logging/logfields"
  "github.com/cilium/cilium/pkg/maps/bwmap"
 )
@@ -28,8 +29,6 @@ const (
  // IngressBandwidth is the K8s Pod annotation.
  IngressBandwidth = "kubernetes.io/ingress-bandwidth"
 
- EnableBBR = "enable-bbr"
-
  // FqDefaultHorizon represents maximum allowed departure
  // time delta in future. Given applications can set SO_TXTIME
  // from user space this is a limit to prevent buggy applications
@@ -69,11 +68,26 @@ func (m *manager) defines() (defines.Map, error) {
  return cDefinesMap, nil
 }
 
-func (m *manager) DeleteEndpointBandwidthLimit(epID uint16) error {
+func (m *manager) UpdateBandwidthLimit(epID uint16, bytesPerSecond uint64) {
  if m.enabled {
- return bwmap.Delete(epID)
+ txn := m.params.DB.WriteTxn(m.params.EdtTable)
+ m.params.EdtTable.Insert(
+ txn,
+ bwmap.NewEdt(epID, bytesPerSecond),
+ )
+ txn.Commit()
+ }
+}
+
+func (m *manager) DeleteBandwidthLimit(epID uint16) {
+ if m.enabled {
+ txn := m.params.DB.WriteTxn(m.params.EdtTable)
+ obj, _, found := m.params.EdtTable.Get(txn, bwmap.EdtIDIndex.Query(epID))
+ if found {
+ m.params.EdtTable.Delete(txn, obj)
+ }
+ txn.Commit()
  }
- return nil
 }
 
 func GetBytesPerSec(bandwidth string) (uint64, error) {
@@ -111,7 +125,7 @@ func (m *manager) probe() error {
  // - https://lpc.events/event/11/contributions/953/
  // - https://lore.kernel.org/bpf/[email protected]/
  if probes.HaveProgramHelper(ebpf.SchedCLS, asm.FnSkbSetTstamp) != nil {
- return fmt.Errorf("cannot enable --%s, needs kernel 5.18 or newer", EnableBBR)
+ return fmt.Errorf("cannot enable --%s, needs kernel 5.18 or newer", types.EnableBBRFlag)
  }
  }
 

pkg/datapath/linux/bandwidth/cell.go

@@ -13,12 +13,12 @@ import (
  "github.com/cilium/statedb"
  "github.com/cilium/statedb/reconciler"
  "github.com/sirupsen/logrus"
- "github.com/spf13/pflag"
 
  "github.com/cilium/cilium/pkg/datapath/linux/config/defines"
  "github.com/cilium/cilium/pkg/datapath/linux/sysctl"
  "github.com/cilium/cilium/pkg/datapath/tables"
  "github.com/cilium/cilium/pkg/datapath/types"
+ "github.com/cilium/cilium/pkg/maps/bwmap"
  "github.com/cilium/cilium/pkg/option"
  "github.com/cilium/cilium/pkg/time"
 )
@@ -27,7 +27,7 @@ var Cell = cell.Module(
  "bandwidth-manager",
  "Linux Bandwidth Manager for EDT-based pacing",
 
- cell.Config(Config{false, false}),
+ cell.Config(types.DefaultBandwidthConfig),
  cell.Provide(newBandwidthManager),
 
  cell.ProvidePrivate(
@@ -37,19 +37,6 @@ var Cell = cell.Module(
  cell.Invoke(registerReconciler),
 )
 
-type Config struct {
- // EnableBandwidthManager enables EDT-based pacing
- EnableBandwidthManager bool
-
- // EnableBBR enables BBR TCP congestion control for the node including Pods
- EnableBBR bool
-}
-
-func (def Config) Flags(flags *pflag.FlagSet) {
- flags.Bool("enable-bandwidth-manager", def.EnableBandwidthManager, "Enable BPF bandwidth manager")
- flags.Bool(EnableBBR, def.EnableBBR, "Enable BBR for the bandwidth manager")
-}
-
 func newReconcilerConfig(log logrus.FieldLogger, tbl statedb.RWTable[*tables.BandwidthQDisc], bwm types.BandwidthManager) reconciler.Config[*tables.BandwidthQDisc] {
  return reconciler.Config[*tables.BandwidthQDisc]{
  Table: tbl,
@@ -93,13 +80,15 @@ type bandwidthManagerParams struct {
  cell.In
 
  Log logrus.FieldLogger
- Config Config
+ Config types.BandwidthConfig
  DaemonConfig *option.DaemonConfig
  Sysctl sysctl.Sysctl
+ DB *statedb.DB
+ EdtTable statedb.RWTable[bwmap.Edt]
 }
 
 func registerReconciler(
- cfg Config,
+ cfg types.BandwidthConfig,
  deriveParams statedb.DeriveParams[*tables.Device, *tables.BandwidthQDisc],
  config reconciler.Config[*tables.BandwidthQDisc],
  reconcilerParams reconciler.Params,

pkg/datapath/linux/datapath.go

@@ -114,10 +114,6 @@ func (l *linuxDatapath) BandwidthManager() datapath.BandwidthManager {
  return l.bwmgr
 }
 
-func (l *linuxDatapath) DeleteEndpointBandwidthLimit(epID uint16) error {
- return l.bwmgr.DeleteEndpointBandwidthLimit(epID)
-}
-
 func (l *linuxDatapath) Orchestrator() datapath.Orchestrator {
  return l.orchestrator
 }

pkg/datapath/types/bandwidth.go

@@ -3,8 +3,35 @@
 
 package types
 
+import "github.com/spf13/pflag"
+
+const (
+ EnableBandwidthManagerFlag = "enable-bandwidth-manager"
+ EnableBBRFlag = "enable-bbr"
+)
+
+type BandwidthConfig struct {
+ // EnableBandwidthManager enables EDT-based pacing
+ EnableBandwidthManager bool
+
+ // EnableBBR enables BBR TCP congestion control for the node including Pods
+ EnableBBR bool
+}
+
+func (def BandwidthConfig) Flags(flags *pflag.FlagSet) {
+ flags.Bool(EnableBandwidthManagerFlag, def.EnableBandwidthManager, "Enable BPF bandwidth manager")
+ flags.Bool(EnableBBRFlag, def.EnableBBR, "Enable BBR for the bandwidth manager")
+}
+
+var DefaultBandwidthConfig = BandwidthConfig{
+ EnableBandwidthManager: false,
+ EnableBBR: false,
+}
+
 type BandwidthManager interface {
  BBREnabled() bool
- DeleteEndpointBandwidthLimit(epID uint16) error
  Enabled() bool
+
+ UpdateBandwidthLimit(endpointID uint16, bytesPerSecond uint64)
+ DeleteBandwidthLimit(endpointID uint16)
 }

pkg/endpoint/bpf.go

@@ -982,9 +982,7 @@ func (e *Endpoint) deleteMaps() []error {
 
  // Remove rate limit from bandwidth manager map.
  if e.bps != 0 {
- if err := e.owner.Datapath().BandwidthManager().DeleteEndpointBandwidthLimit(e.ID); err != nil {
- errors = append(errors, fmt.Errorf("removing endpoint from bandwidth manager map: %w", err))
- }
+ e.owner.Datapath().BandwidthManager().DeleteBandwidthLimit(e.ID)
  }
 
  if e.ConntrackLocalLocked() {

pkg/endpoint/events.go

@@ -15,7 +15,6 @@ import (
  datapath "github.com/cilium/cilium/pkg/datapath/types"
  "github.com/cilium/cilium/pkg/eventqueue"
  "github.com/cilium/cilium/pkg/logging/logfields"
- "github.com/cilium/cilium/pkg/maps/bwmap"
  "github.com/cilium/cilium/pkg/option"
  "github.com/cilium/cilium/pkg/policy"
 )
@@ -300,10 +299,12 @@ func (ev *EndpointPolicyBandwidthEvent) Handle(res chan interface{}) {
  if bandwidthEgress != "" {
  bps, err = bandwidth.GetBytesPerSec(bandwidthEgress)
  if err == nil {
- err = bwmap.Update(e.ID, bps)
+ ev.bwm.UpdateBandwidthLimit(e.ID, bps)
+ } else {
+ e.getLogger().WithError(err).Debugf("failed to parse bandwidth limit %q", bandwidthEgress)
  }
  } else {
- err = bwmap.SilentDelete(e.ID)
+ ev.bwm.DeleteBandwidthLimit(e.ID)
  }
  if err != nil {
  res <- &EndpointRegenerationResult{

pkg/maps/bwmap/bwmap.go

@@ -5,12 +5,13 @@ package bwmap
 
 import (
  "fmt"
- "sync"
+
+ "github.com/cilium/hive/cell"
 
  "github.com/cilium/cilium/pkg/bpf"
+ "github.com/cilium/cilium/pkg/datapath/types"
  "github.com/cilium/cilium/pkg/ebpf"
  "github.com/cilium/cilium/pkg/maps/lxcmap"
- "github.com/cilium/cilium/pkg/option"
  "github.com/cilium/cilium/pkg/time"
 )
 
@@ -43,41 +44,35 @@ type EdtInfo struct {
 func (v *EdtInfo) String() string { return fmt.Sprintf("%d", int(v.Bps)) }
 func (v *EdtInfo) New() bpf.MapValue { return &EdtInfo{} }
 
-var (
- throttleMap *bpf.Map
- throttleMapInit = &sync.Once{}
-)
-
-func ThrottleMap() *bpf.Map {
- throttleMapInit.Do(func() {
- throttleMap = bpf.NewMap(
- MapName,
- ebpf.Hash,
- &EdtId{},
- &EdtInfo{},
- MapSize,
- bpf.BPF_F_NO_PREALLOC,
- ).WithCache().WithPressureMetric().
- WithEvents(option.Config.GetEventBufferConfig(MapName))
- })
-
- return throttleMap
+type throttleMap struct {
+ *bpf.Map
 }
 
-func Update(Id uint16, Bps uint64) error {
- return ThrottleMap().Update(
- &EdtId{Id: uint64(Id)},
- &EdtInfo{Bps: Bps, TimeHorizonDrop: uint64(DefaultDropHorizon)})
-}
-
-func Delete(Id uint16) error {
- return ThrottleMap().Delete(
- &EdtId{Id: uint64(Id)})
+// ThrottleMap constructs the cilium_throttle map. Direct use of this
+// outside of this package is solely for cilium-dbg.
+func ThrottleMap() *bpf.Map {
+ return bpf.NewMap(
+ MapName,
+ ebpf.Hash,
+ &EdtId{},
+ &EdtInfo{},
+ MapSize,
+ bpf.BPF_F_NO_PREALLOC,
+ )
 }
 
-func SilentDelete(Id uint16) error {
- _, err := ThrottleMap().SilentDelete(
- &EdtId{Id: uint64(Id)})
-
- return err
+func newThrottleMap(cfg types.BandwidthConfig, lc cell.Lifecycle) (out bpf.MapOut[throttleMap]) {
+ m := throttleMap{ThrottleMap()}
+ if cfg.EnableBandwidthManager {
+ // Only open the map if bandwidth manager is enabled.
+ lc.Append(cell.Hook{
+ OnStart: func(cell.HookContext) error {
+ return m.OpenOrCreate()
+ },
+ OnStop: func(cell.HookContext) error {
+ return m.Close()
+ },
+ })
+ }
+ return bpf.NewMapOut(m)
 }