first commit

Author: mazdakn

felix/bpf/proxy/proxy_test.go

@@ -610,6 +610,83 @@ var _ = Describe("BPF Proxy", func() {
 			})
 		})
 	})
+
+	Context("with terminating workloads", func() {
+		var (
+			p   proxy.Proxy
+			dp  *mockSyncer
+			k8s *fake.Clientset
+		)
+
+		BeforeEach(func() {
+			By("creating proxy with fake client and mock syncer", func() {
+				var err error
+
+				k8s = fake.NewSimpleClientset(testSvc, testSvcEpsSlice)
+				syncStop = make(chan struct{})
+				dp = newMockSyncer(syncStop)
+
+				opts := []proxy.Option{proxy.WithImmediateSync()}
+
+				p, err = proxy.New(k8s, dp, "test-node", opts...)
+				Expect(err).NotTo(HaveOccurred())
+			})
+		})
+
+		AfterEach(func() {
+			By("stopping the proxy", func() {
+				close(syncStop)
+				p.Stop()
+			})
+		})
+
+		It("should see IsReady=false and IsTerminating=true", func() {
+			By("getting the initial sync")
+
+			dp.checkState(func(s proxy.DPSyncerState) {
+				Expect(len(s.SvcMap)).To(Equal(1))
+				Expect(len(s.EpsMap)).To(Equal(1))
+
+			})
+
+			By("placing one endpoint to terminating state")
+
+			falsePtr := new(bool)
+			*falsePtr = false
+
+			truePtr := new(bool)
+			*truePtr = true
+
+			testSvcEpsSlice.Endpoints[0].Conditions.Ready = falsePtr
+			testSvcEpsSlice.Endpoints[0].Conditions.Terminating = truePtr
+			err := k8s.Tracker().Update(discovery.SchemeGroupVersion.WithResource("endpointslices"),
+				testSvcEpsSlice, "default")
+			Expect(err).NotTo(HaveOccurred())
+
+			dp.checkState(func(s proxy.DPSyncerState) {
+				Expect(len(s.SvcMap)).To(Equal(1))
+				Expect(len(s.EpsMap)).To(Equal(1))
+
+				var key k8sp.ServicePortName
+
+				for key = range s.EpsMap {
+				}
+
+				isReady := 0
+				isTerminating := 0
+				for _, ep := range s.EpsMap[key] {
+					if ep.IsReady() {
+						isReady++
+					}
+					if ep.IsTerminating() {
+						isTerminating++
+					}
+				}
+				Expect(isReady).To(Equal(1))
+				Expect(isTerminating).To(Equal(1))
+			})
+		})
+	})
 })
 
 type mockSyncer struct {

felix/calc/calc_graph.go

@@ -16,6 +16,7 @@ package calc
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
 	log "github.com/sirupsen/logrus"
 
 	v3 "github.com/projectcalico/api/pkg/apis/projectcalico/v3"
@@ -378,7 +379,8 @@ func NewCalculationGraph(callbacks PipelineCallbacks, conf *config.Config, liveC
 	hostIPPassthru.RegisterWith(allUpdDispatcher)
 	cg.hostIPPassthru = hostIPPassthru
 
-	if conf.BPFEnabled || conf.Encapsulation.VXLANEnabled || conf.Encapsulation.VXLANEnabledV6 || conf.WireguardEnabled || conf.WireguardEnabledV6 {
+	if conf.BPFEnabled || conf.Encapsulation.VXLANEnabled || conf.Encapsulation.VXLANEnabledV6 ||
+		conf.WireguardEnabled || conf.WireguardEnabledV6 || conf.Encapsulation.IPIPEnabled {
 		// Calculate simple node-ownership routes.
 		//        ...
 		//     Dispatcher (all updates)
@@ -391,6 +393,7 @@ func NewCalculationGraph(callbacks PipelineCallbacks, conf *config.Config, liveC
 		//         |
 		//      <dataplane>
 		//
+		logrus.Info("Marmar")
 		l3RR := NewL3RouteResolver(hostname, callbacks, conf.UseNodeResourceUpdates(), conf.RouteSource)
 		l3RR.RegisterWith(allUpdDispatcher, localEndpointDispatcher)
 		l3RR.OnAlive = liveCallback

felix/dataplane/linux/bpf_ep_mgr.go

@@ -3405,6 +3405,7 @@ func (m *bpfEndpointManager) loadPolicyProgram(
 	log.WithFields(log.Fields{
 		"progName": progName,
 		"ipFamily": ipFamily,
+		"rules":    rules,
 	}).Debug("Generating policy program...")
 
 	ipsetsMapFD := m.v4.IpsetsMap.MapFD()

felix/dataplane/linux/int_dataplane.go

@@ -804,8 +804,29 @@ func NewIntDataplaneDriver(config Config) *InternalDataplane {
 	dp.RegisterManager(newFloatingIPManager(natTableV4, ruleRenderer, 4, config.FloatingIPsEnabled))
 	dp.RegisterManager(newMasqManager(ipSetsV4, natTableV4, ruleRenderer, config.MaxIPSetSize, 4))
 	if config.RulesConfig.IPIPEnabled {
+		var routeTableIPIP routetable.RouteTableInterface
+		if !config.RouteSyncDisabled {
+			log.Debug("RouteSyncDisabled is false.")
+			routeTableIPIP = routetable.New(
+				[]string{"^tunl0$"}, 4, config.NetlinkTimeout,
+				config.DeviceRouteSourceAddress, config.DeviceRouteProtocol, config.RemoveExternalRoutes,
+				unix.RT_TABLE_MAIN, dp.loopSummarizer, featureDetector, routetable.WithLivenessCB(dp.reportHealth),
+				routetable.WithRouteCleanupGracePeriod(routeCleanupGracePeriod))
+		} else {
+			log.Info("RouteSyncDisabled is true, using DummyTable.")
+			routeTableIPIP = &routetable.DummyTable{}
+		}
+
 		// Add a manager to keep the all-hosts IP set up to date.
-		dp.ipipManager = newIPIPManager(ipSetsV4, config.MaxIPSetSize, config.ExternalNodesCidrs)
+		dp.ipipManager = newIPIPManager(
+			ipSetsV4,
+			"tunl0",
+			routeTableIPIP,
+			config,
+			dp.loopSummarizer,
+			4,
+			featureDetector,
+		)
 		dp.RegisterManager(dp.ipipManager) // IPv4-only
 	} else {
 		// Only clean up IPIP addresses if IPIP is implicitly disabled (no IPIP pools and not explicitly set in FelixConfig)