labels and annotations filters

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/docker/docker v20.10.1+incompatible // indirect
 	github.com/docker/go-connections v0.4.0
 	github.com/go-logr/logr v0.3.0 // indirect
+	github.com/gobwas/glob v0.2.3
 	github.com/gogo/protobuf v1.3.1
 	github.com/golang/protobuf v1.4.3
 	github.com/google/btree v1.0.0

go.sum

@@ -120,6 +120,8 @@ github.com/go-openapi/jsonreference v0.0.0-20160704190145-13c6e3589ad9/go.mod h1
 github.com/go-openapi/spec v0.0.0-20160808142527-6aced65f8501/go.mod h1:J8+jY1nAiCcj+friV/PDoE1/3eeccG9LYBs0tYvLOWc=
 github.com/go-openapi/swag v0.0.0-20160704191624-1d0bd113de87/go.mod h1:DXUve3Dpr1UfpPtxFw+EFuQ41HhCWZfha5jSVRG7C7I=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y=
+github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
 github.com/gogo/protobuf v1.3.1 h1:DqDEcV5aeaTmdFBePNpYsp3FlcVH/2ISVVM9Qf8PSls=

pkg/api/localnetv1/services.proto

@@ -140,6 +140,7 @@ message NodeInfo {
 message Node {
     string Name = 1;
     map<string, string> Labels = 2;
+    map<string, string> Annotations = 3;
 }
 
 service Global {

pkg/endpoints/node-event-handler.go

@@ -12,28 +12,40 @@ import (
 
 type nodeEventHandler struct{ eventHandler }
 
-var myNodeName = flag.String("node-name", "", "Node name override")
+var (
+	myNodeName = flag.String("node-name", "", "Node name override")
+
+	nodeLabelGlobs      = flag.String("with-node-labels", "", "node labels to include")
+	nodeAnnotationGlobs = flag.String("with-node-annotations", "", "node annotations to include")
+)
 
 func (h nodeEventHandler) OnAdd(obj interface{}) {
 	node := obj.(*v1.Node)
 
 	// keep only what we want
 	n := &localnetv1.Node{
-		Name:   node.Name,
-		Labels: node.Labels,
+		Name:        node.Name,
+		Labels:      globsFilter(node.Labels, *nodeLabelGlobs),
+		Annotations: globsFilter(node.Annotations, *nodeAnnotationGlobs),
 	}
 
 	h.s.Update(func(tx *proxystore.Tx) {
 		tx.SetNode(n)
 
 		if !proxy.ManageEndpointSlices {
 			// endpoints => need to update all matching topologies
+			toSet := make([]*localnetv1.EndpointInfo, 0)
 			tx.Each(proxystore.Endpoints, func(kv *proxystore.KV) bool {
 				if kv.Endpoint.NodeName == n.Name {
 					kv.Endpoint.Topology = n.Labels
+					toSet = append(toSet, kv.Endpoint)
 				}
 				return true
 			})
+
+			for _, ei := range toSet {
+				tx.SetEndpoint(ei)
+			}
 		}
 
 		h.updateSync(proxystore.Nodes, tx)

pkg/endpoints/service-event-handler.go

@@ -1,11 +1,20 @@
 package endpoints
 
 import (
-	"m.cluseau.fr/kube-proxy2/pkg/api/localnetv1"
-	"m.cluseau.fr/kube-proxy2/pkg/proxystore"
+	"flag"
+	"strings"
+
+	"github.com/gobwas/glob"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/klog"
+	"m.cluseau.fr/kube-proxy2/pkg/api/localnetv1"
+	"m.cluseau.fr/kube-proxy2/pkg/proxystore"
+)
+
+var (
+	serviceLabelGlobs      = flag.String("with-service-labels", "", "service labels to include")
+	serviceAnnonationGlobs = flag.String("with-service-annotations", "", "service annotations to include")
 )
 
 type serviceEventHandler struct{ eventHandler }
@@ -17,8 +26,8 @@ func (h *serviceEventHandler) OnAdd(obj interface{}) {
 		Namespace:   svc.Namespace,
 		Name:        svc.Name,
 		Type:        string(svc.Spec.Type),
-		Labels:      svc.Labels,
-		Annotations: svc.Annotations,
+		Labels:      globsFilter(svc.Labels, *serviceLabelGlobs),
+		Annotations: globsFilter(svc.Annotations, *serviceAnnonationGlobs),
 		MapIP:       false, // TODO for headless? or no ports means all? why am I adding those questions? ;-)
 		IPs: &localnetv1.ServiceIPs{
 			ClusterIP:   svc.Spec.ClusterIP,
@@ -94,3 +103,32 @@ func (h *serviceEventHandler) OnDelete(oldObj interface{}) {
 		h.updateSync(proxystore.Services, tx)
 	})
 }
+
+// FIXME move to a common place
+func globsFilter(src map[string]string, globsFlag string) (dst map[string]string) {
+	if len(globsFlag) == 0 {
+		return
+	}
+
+	globsS := strings.Split(globsFlag, ",")
+
+	globs := make([]glob.Glob, len(globsS))
+
+	for i, globS := range globsS {
+		globs[i] = glob.MustCompile(globS)
+	}
+
+	dst = make(map[string]string)
+
+srcLoop:
+	for k, v := range src {
+		for _, g := range globs {
+			if g.Match(k) {
+				dst[k] = v
+				continue srcLoop
+			}
+		}
+	}
+
+	return
+}

pkg/proxystore/proxystore.go

@@ -370,6 +370,63 @@ func (tx *Tx) DelEndpointsOfSource(namespace, sourceName string) {
 	}
 }
 
+func (tx *Tx) SetEndpoint(ei *localnetv1.EndpointInfo) {
+	tx.roPanic()
+
+	newHash := tx.s.hashOf(&localnetv1.EndpointInfo{
+		Endpoint:   ei.Endpoint,
+		Conditions: ei.Conditions,
+		Topology:   ei.Topology,
+	})
+
+	if ei.Hash == newHash {
+		return // not changed
+	}
+
+	prevKey := strconv.FormatUint(ei.Hash, 16)
+
+	tx.del(&KV{
+		Set:       Endpoints,
+		Namespace: ei.Namespace,
+		Name:      ei.ServiceName,
+		Source:    ei.SourceName,
+		Key:       prevKey,
+	})
+
+	// also delete by source only
+	tx.del(&KV{
+		Set:       Endpoints,
+		Namespace: ei.Namespace,
+		Source:    ei.SourceName,
+		Key:       prevKey,
+	})
+
+	// update key
+	ei.Hash = newHash
+	key := strconv.FormatUint(ei.Hash, 16)
+
+	tx.set(&KV{
+		Set:       Endpoints,
+		Namespace: ei.Namespace,
+		Name:      ei.ServiceName,
+		Source:    ei.SourceName,
+		Key:       key,
+		Value:     ei,
+		Endpoint:  ei,
+	})
+
+	// also index by source only
+	tx.set(&KV{
+		Set:       Endpoints,
+		Namespace: ei.Namespace,
+		Source:    ei.SourceName,
+		Key:       key,
+		Value:     ei,
+		Endpoint:  ei,
+	})
+
+}
+
 // Nodes funcs
 
 func (tx *Tx) GetNode(name string) *localnetv1.Node {