Experimental load-balancing control-plane with StateDB

Documentation/spelling_wordlist.txt

@@ -329,6 +329,7 @@ fromCIDR
 fromCIDRSet
 fromEndpoints
 frontend
+frontends
 fs
 func
 gRPC

cilium-dbg/cmd/statedb.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cilium/statedb"
 
 	clientPkg "github.com/cilium/cilium/pkg/client"
+	"github.com/cilium/cilium/pkg/loadbalancer/experimental"
 
 	"github.com/cilium/cilium/pkg/datapath/tables"
 	"github.com/cilium/cilium/pkg/healthv2"
@@ -49,6 +50,11 @@ var statedbDumpCmd = &cobra.Command{
 	},
 }
 
+var statedbExperimentalCmd = &cobra.Command{
+	Use:   "experimental",
+	Short: "Experimental",
+}
+
 // StateDB HTTP handler is mounted at /statedb by configureAPIServer() in daemon/cmd/cells.go.
 var statedbURL, _ = url.Parse("http://localhost/statedb")
 
@@ -127,8 +133,14 @@ func statedbTableCommand[Obj statedb.TableWritable](tableName string) *cobra.Com
 }
 
 func init() {
+	statedbExperimentalCmd.AddCommand(
+		statedbTableCommand[*experimental.Service](experimental.ServiceTableName),
+		statedbTableCommand[*experimental.Frontend](experimental.FrontendTableName),
+		statedbTableCommand[*experimental.Backend](experimental.BackendTableName),
+	)
 	StatedbCmd.AddCommand(
 		statedbDumpCmd,
+		statedbExperimentalCmd,
 
 		statedbTableCommand[*tables.Device]("devices"),
 		statedbTableCommand[*tables.Route]("routes"),

daemon/cmd/cells.go

@@ -37,6 +37,7 @@ import (
 	k8sSynced "github.com/cilium/cilium/pkg/k8s/synced"
 	"github.com/cilium/cilium/pkg/kvstore/store"
 	"github.com/cilium/cilium/pkg/l2announcer"
+	loadbalancer_experimental "github.com/cilium/cilium/pkg/loadbalancer/experimental"
 	"github.com/cilium/cilium/pkg/logging/logfields"
 	"github.com/cilium/cilium/pkg/maps/metricsmap"
 	"github.com/cilium/cilium/pkg/metrics"
@@ -163,6 +164,9 @@ var (
 		// daemonCell wraps the legacy daemon initialization and provides Promise[*Daemon].
 		daemonCell,
 
+		// Experimental control-plane for configuring service load-balancing.
+		loadbalancer_experimental.Cell,
+
 		// Service is a datapath service handler. Its main responsibility is to reflect
 		// service-related changes into BPF maps used by datapath BPF programs.
 		service.Cell,

pkg/clustermesh/types/addressing.go

@@ -4,6 +4,8 @@
 package types
 
 import (
+	"bytes"
+	"errors"
 	"fmt"
 	"net"
 	"net/netip"
@@ -42,6 +44,42 @@ type AddrCluster struct {
 
 const AddrClusterLen = 20
 
+// MarshalJSON marshals the address as a string in the form
+// <addr>@<clusterID>, e.g. "1.2.3.4@1"
+func (a AddrCluster) MarshalJSON() ([]byte, error) {
+	var b bytes.Buffer
+	b.WriteByte('"')
+	if !a.addr.IsValid() {
+		b.WriteString("0.0.0.0")
+	} else {
+		b.WriteString(a.addr.String())
+	}
+	if a.clusterID != 0 {
+		b.WriteByte('@')
+		b.WriteString(
+			strconv.FormatUint(uint64(a.clusterID), 10),
+		)
+	}
+	b.WriteByte('"')
+	return b.Bytes(), nil
+}
+
+func (a *AddrCluster) UnmarshalJSON(data []byte) error {
+	if len(data) == 0 || data[0] != '"' || data[len(data)-1] != '"' {
+		return errors.New("AddrCluster.UnmarshalJSON: bad address")
+	}
+	// Drop the parens
+	data = data[1 : len(data)-1]
+
+	a2, err := ParseAddrCluster(string(data))
+	if err != nil {
+		return err
+	}
+	a.addr = a2.addr
+	a.clusterID = a2.clusterID
+	return nil
+}
+
 // ParseAddrCluster parses s as an IP + ClusterID and returns AddrCluster.
 // The string s can be a bare IP string (any IP address format allowed in
 // netip.ParseAddr()) or IP string + @ + ClusterID with decimal. Bare IP

pkg/container/immset.go

@@ -5,6 +5,7 @@ package container
 
 import (
 	"cmp"
+	"encoding/json"
 	"slices"
 
 	"golang.org/x/exp/constraints"
@@ -44,6 +45,14 @@ func (s ImmSet[T]) Has(x T) bool {
 	return found
 }
 
+func (s *ImmSet[T]) MarshalJSON() ([]byte, error) {
+	return json.Marshal(s.xs)
+}
+
+func (s *ImmSet[T]) UnmarshalJSON(data []byte) error {
+	return json.Unmarshal(data, &s.xs)
+}
+
 func (s ImmSet[T]) Insert(xs ...T) ImmSet[T] {
 	xs2 := make([]T, 0, len(s.xs)+len(xs))
 	xs2 = append(xs2, s.xs...)

pkg/loadbalancer/experimental/backend.go

@@ -0,0 +1,116 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Authors of Cilium
+
+package experimental
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/cilium/statedb"
+	"github.com/cilium/statedb/index"
+
+	"github.com/cilium/cilium/pkg/container"
+	"github.com/cilium/cilium/pkg/loadbalancer"
+	"github.com/cilium/cilium/pkg/source"
+)
+
+const (
+	BackendTableName = "backends"
+)
+
+// Backend describes a backend for one or more services.
+//
+// This embeds the loadbalancer.Backend and adds further fields to it. Eventually
+// we'd unify them, but for now we do not want to affect existing code yet.
+type Backend struct {
+	loadbalancer.Backend
+
+	// Cluster in which the backend resides
+	Cluster string
+
+	// Source of the backend
+	Source source.Source
+
+	// ReferencedBy is the set of services referencing this backend.
+	// This is managed by [Services].
+	ReferencedBy container.ImmSet[loadbalancer.ServiceName]
+}
+
+func (be *Backend) String() string {
+	return strings.Join(be.TableRow(), " ")
+}
+
+func (be *Backend) TableHeader() []string {
+	return []string{
+		"Address",
+		"State",
+		"Source",
+		"ReferencedBy",
+	}
+}
+
+func toStrings[T fmt.Stringer](xs []T) []string {
+	out := make([]string, len(xs))
+	for i := range xs {
+		out[i] = xs[i].String()
+	}
+	return out
+}
+
+func (be *Backend) TableRow() []string {
+	state, err := be.State.String()
+	if err != nil {
+		state = err.Error()
+	}
+	return []string{
+		be.L3n4Addr.StringWithProtocol(),
+		state,
+		string(be.Source),
+		strings.Join(toStrings(be.ReferencedBy.AsSlice()), ", "),
+	}
+}
+
+func (be *Backend) release(name loadbalancer.ServiceName) (*Backend, bool) {
+	beCopy := *be
+	beCopy.ReferencedBy = beCopy.ReferencedBy.Delete(name)
+	return &beCopy, beCopy.ReferencedBy.Len() == 0
+}
+
+// Clone returns a shallow clone of the backend.
+func (be *Backend) Clone() *Backend {
+	be2 := *be
+	return &be2
+}
+
+var (
+	BackendAddrIndex = statedb.Index[*Backend, loadbalancer.L3n4Addr]{
+		Name: "addr",
+		FromObject: func(obj *Backend) index.KeySet {
+			return index.NewKeySet(l3n4AddrKey(obj.L3n4Addr))
+		},
+		FromKey: l3n4AddrKey,
+		Unique:  true,
+	}
+
+	BackendServiceIndex = statedb.Index[*Backend, loadbalancer.ServiceName]{
+		Name: "service-name",
+		FromObject: func(obj *Backend) index.KeySet {
+			return index.StringerSlice(obj.ReferencedBy.AsSlice())
+		},
+		FromKey: index.Stringer[loadbalancer.ServiceName],
+		Unique:  false,
+	}
+)
+
+func NewBackendsTable(db *statedb.DB) (statedb.RWTable[*Backend], error) {
+	tbl, err := statedb.NewTable(
+		BackendTableName,
+		BackendAddrIndex,
+		BackendServiceIndex,
+	)
+	if err != nil {
+		return nil, err
+	}
+	return tbl, db.RegisterTable(tbl)
+}