Merge pull request kubernetes-retired#314 from gauravkghildiyal/ipvs

Change endpoint key and use anonymous set depending on presence of PodName

Author: k8s-ci-robot

api/localnetv1/services.pb.go

@@ -162,6 +162,7 @@ message EndpointInfo {
     string SourceName = 3;
     string ServiceName = 4;
     string NodeName = 5;
+    string PodName = 9;
 
     Endpoint Endpoint = 6;
     EndpointConditions Conditions = 7;

api/localnetv1/services.proto

@@ -58,7 +58,6 @@ type SessionAffinityListener interface {
 // - AddIPPort
 // - DeleteIPPort
 // - DeleteIP
-//
 type ServicesListener struct {
 	PortsListener           PortsListener
 	IPsListener             IPsListener
@@ -72,7 +71,6 @@ type ServicesListener struct {
 // New creates a new ServicesListener.
 //
 // Reminder: you need to associate listeners for this listener to be useful.
-//
 func New() *ServicesListener {
 	return &ServicesListener{
 		services: map[string]*localnetv1.Service{},
@@ -251,7 +249,12 @@ func (sl *ServicesListener) diff(prevSvc, currSvc *localnetv1.Service) {
 }
 
 func samePort(p1, p2 *localnetv1.PortMapping) bool {
-	return p1.Protocol == p2.Protocol && p1.Port == p2.Port
+	return p1.Name == p2.Name &&
+		p1.Protocol == p2.Protocol &&
+		p1.Port == p2.Port &&
+		p1.NodePort == p2.NodePort &&
+		p1.TargetPort == p2.TargetPort &&
+		p1.TargetPortName == p2.TargetPortName
 }
 
 // SessionAffinity contains data about assinged session affinity

api/localnetv1/services_grpc.pb.go

@@ -70,6 +70,10 @@ func (h *endpointsEventHandler) OnAdd(obj interface{}) {
 						}
 					}
 
+					if t := addr.TargetRef; t != nil && t.Kind == "Pod" {
+						info.PodName = t.Name
+					}
+
 					if addr.IP != "" {
 						info.Endpoint.AddAddress(addr.IP)
 					}

client/serviceevents/service-events.go

@@ -60,6 +60,10 @@ func (h sliceEventHandler) OnAdd(obj interface{}) {
 			info.NodeName = sliceEndpoint.Topology[hostNameLabel]
 		}
 
+		if t := sliceEndpoint.TargetRef; t != nil && t.Kind == "Pod" {
+			info.PodName = t.Name
+		}
+
 		if h := sliceEndpoint.Hostname; h != nil {
 			info.Endpoint.Hostname = *h
 		}

server/jobs/kube2store/endpoints-event-handler.go

@@ -46,6 +46,7 @@ func (j *Job) Run(ctx context.Context) error {
 		Sets: []localnetv1.Set{
 			localnetv1.Set_ServicesSet,
 			localnetv1.Set_EndpointsSet,
+			localnetv1.Set_EndpointsSet, // 2nd endpoints set for endpoints which do not have a corresponding pod name
 		},
 		Sink: run,
 	}
@@ -77,6 +78,7 @@ func (s *jobRun) Update(tx *proxystore.Tx, w *watchstate.WatchState) {
 
 	svcs := w.StoreFor(localnetv1.Set_ServicesSet)
 	seps := w.StoreFor(localnetv1.Set_EndpointsSet)
+	sepsAnonymous := w.StoreForN(localnetv1.Set_EndpointsSet, 1)
 
 	// set all new values
 	tx.Each(proxystore.Services, func(kv *proxystore.KV) bool {
@@ -94,30 +96,44 @@ func (s *jobRun) Update(tx *proxystore.Tx, w *watchstate.WatchState) {
 			// hash only the endpoint
 			hash := serde.Hash(ei.Endpoint)
 
-			// key is service key + endpoint hash (64 bits, in hex)
-			key := append(make([]byte, 0, len(key)+1+64/8*2), key...)
-			key = append(key, '/')
-			key = strconv.AppendUint(key, hash, 16)
+			var epKey []byte
+			var set *lightdiffstore.DiffStore
+
+			if ei.PodName == "" {
+				set = sepsAnonymous
+				// key is service key + endpoint hash (64 bits, in hex)
+				epKey = append(make([]byte, 0, len(key)+1+64/8*2), key...)
+				epKey = append(epKey, '/')
+				epKey = strconv.AppendUint(epKey, hash, 16)
+			} else {
+				set = seps
+				// key is service key + podName
+				epKey = append(make([]byte, 0, len(key)+1+len(ei.PodName)), key...)
+				epKey = append(epKey, '/')
+				epKey = append(epKey, []byte(ei.PodName)...)
+			}
 
 			if trace.IsEnabled() {
-				trace.Log(ctx, "endpoint", string(key))
+				trace.Log(ctx, "endpoint", string(epKey))
 			}
 
-			seps.Set(key, hash, ei.Endpoint)
+			set.Set(epKey, hash, ei.Endpoint)
 		}
 
 		return true
 	})
 }
 
-func (_ *jobRun) SendDiff(w *watchstate.WatchState) (updated bool) {
+func (*jobRun) SendDiff(w *watchstate.WatchState) (updated bool) {
 	_, task := trace.NewTask(context.Background(), "LocalState.SendDiff")
 	defer task.End()
 
 	count := 0
 	count += w.SendUpdates(localnetv1.Set_ServicesSet)
+	count += w.SendDeletesN(localnetv1.Set_EndpointsSet, 1)
 	count += w.SendUpdates(localnetv1.Set_EndpointsSet)
 	count += w.SendDeletes(localnetv1.Set_EndpointsSet)
+	count += w.SendUpdatesN(localnetv1.Set_EndpointsSet, 1)
 	count += w.SendDeletes(localnetv1.Set_ServicesSet)
 
 	w.Reset(lightdiffstore.ItemDeleted)