weave/lb/egress_api.go

package lb

import (
	"context"
	"encoding/json"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/runtime/serializer"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/rest"
)

// EgressService (reweave.levk.dev/v1alpha1) makes the pods a LoadBalancer Service selects leave the cluster
// from the Service's address. It is named after the Service (same namespace, same name). Its semantics follow
// OVN-Kubernetes' EgressService (k8s.ovn.org/v1, sourceIPBy: LoadBalancerIP) where they agree; they differ in
// that membership is every pod the Service's selector matches (by pod IP, never by readiness), the holder is
// the node that holds the address (no election), and a member's traffic to the outside leaves as the address
// or not at all.
type EgressService struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`
	Spec              EgressServiceSpec   `json:"spec"`
	Status            EgressServiceStatus `json:"status,omitempty"`
}

// EgressServiceSpec is what the operator asks for.
type EgressServiceSpec struct {
	// SourceIPBy is LoadBalancerIP: members leave from the Service's load-balancer address (the only mode in v1).
	SourceIPBy string `json:"sourceIPBy,omitempty"`
	// Emit is where members' packets leave the cluster. Explicit, no default: Holder (tunnelled to the node
	// holding the address, which emits them) is the only mode in v1.
	Emit string `json:"emit"`
}

// EgressServiceStatus is written by the holder alone.
type EgressServiceStatus struct {
	// Host is the node that holds the address and emits members' packets.
	Host string `json:"host,omitempty"`
	// Address is the address members leave from.
	Address string `json:"address,omitempty"`
	// Blocks gives every node its own source ports on Address, so the holder routes replies back by port.
	Blocks []PortBlock `json:"blocks,omitempty"`
}

// PortBlock is the source-port range [From, To] on the address that one node's members use.
type PortBlock struct {
	Node string `json:"node"`
	From int32  `json:"from"`
	To   int32  `json:"to"`
}

// EgressServiceList is a list of EgressServices.
type EgressServiceList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items           []EgressService `json:"items"`
}

const (
	// Group and Version of the EgressService API.
	Group   = "reweave.levk.dev"
	Version = "v1alpha1"
	// EgressServices is the resource name.
	EgressServices = "egressservices"

	SourceIPByLoadBalancerIP = "LoadBalancerIP"
	EmitHolder               = "Holder"
)

// GroupVersion of the EgressService API.
var GroupVersion = schema.GroupVersion{Group: Group, Version: Version}

// DeepCopyObject implements runtime.Object.
func (e *EgressService) DeepCopyObject() runtime.Object { return e.DeepCopy() }

// DeepCopy copies the whole object.
func (e *EgressService) DeepCopy() *EgressService {
	if e == nil {
		return nil
	}
	out := *e
	e.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
	out.Status.Blocks = append([]PortBlock(nil), e.Status.Blocks...)
	return &out
}

// DeepCopyObject implements runtime.Object.
func (l *EgressServiceList) DeepCopyObject() runtime.Object {
	if l == nil {
		return nil
	}
	out := *l
	l.ListMeta.DeepCopyInto(&out.ListMeta)
	out.Items = make([]EgressService, len(l.Items))
	for i := range l.Items {
		out.Items[i] = *l.Items[i].DeepCopy()
	}
	return &out
}

func addKnownTypes(s *runtime.Scheme) error {
	s.AddKnownTypes(GroupVersion, &EgressService{}, &EgressServiceList{})
	metav1.AddToGroupVersion(s, GroupVersion)
	return nil
}

// NewEgressClient is a REST client for the EgressService API (no generated clientset: two plain types).
func NewEgressClient(base *rest.Config) (*rest.RESTClient, error) {
	scheme := runtime.NewScheme()
	if err := addKnownTypes(scheme); err != nil {
		return nil, err
	}
	cfg := *base
	cfg.GroupVersion = &GroupVersion
	cfg.APIPath = "/apis"
	cfg.ContentType = runtime.ContentTypeJSON
	cfg.NegotiatedSerializer = serializer.NewCodecFactory(scheme).WithoutConversion()
	if cfg.UserAgent == "" {
		cfg.UserAgent = rest.DefaultKubernetesUserAgent()
	}
	return rest.RESTClientFor(&cfg)
}

// PatchEgressStatus merge-patches an EgressService's status (nil fields are removed).
func PatchEgressStatus(ctx context.Context, c rest.Interface, namespace, name string, st EgressServiceStatus) error {
	body := map[string]interface{}{"host": nil, "address": nil, "blocks": nil}
	if st.Host != "" {
		body["host"] = st.Host
	}
	if st.Address != "" {
		body["address"] = st.Address
	}
	if len(st.Blocks) > 0 {
		body["blocks"] = st.Blocks
	}
	patch, err := json.Marshal(map[string]interface{}{"status": body})
	if err != nil {
		return err
	}
	return c.Patch(types.MergePatchType).Namespace(namespace).Resource(EgressServices).Name(name).
		SubResource("status").Body(patch).Do(ctx).Error()
}