Track PodIP->Namespace mappings, implement endpoint filtering
@@ -393,6 +393,11 @@ func (oc *OvsController) StartNode(mtu uint) error {
}
oc.flowController.AddServiceOFRules(netid, svc.IP, svc.Protocol, svc.Port)
}
+
+ _, err = oc.watchAndGetResource("Pod")
+ if err != nil {
+ return err
+ }
}
if oc.ready != nil {
@@ -543,6 +548,15 @@ func (oc *OvsController) watchServices(ready chan<- bool, start <-chan string) {
}
}
+func (oc *OvsController) watchPods(ready chan<- bool, start <-chan string) {
+ stop := make(chan bool)
+ go oc.subnetRegistry.WatchPods(ready, start, stop)
+
+ <-oc.sig
+ log.Error("Signal received. Stopping watching of pods.")
+ stop <- true
+}
+
func (oc *OvsController) watchCluster(ready chan<- bool, start <-chan string) {
stop := make(chan bool)
clusterEvent := make(chan *api.SubnetEvent)
@@ -605,7 +619,7 @@ func waitForWatchReadiness(ready chan bool, resourceName string) {
// watchAndGetResource will fetch current items in etcd and watch for any new
// changes for the given resource.
-// Supported resources: nodes, subnets, namespaces, services and netnamespaces.
+// Supported resources: nodes, subnets, namespaces, services, netnamespaces, and pods.
//
// To avoid any potential race conditions during this process, these steps are followed:
// 1. Initiator(master/node): Watch for a resource as an async op, lets say WatchProcess
@@ -647,6 +661,10 @@ func (oc *OvsController) watchAndGetResource(resourceName string) (interface{},
go oc.watchServices(ready, start)
waitForWatchReadiness(ready, resourceName)
getOutput, version, err = oc.subnetRegistry.GetServices()
+ case "pod":
+ go oc.watchPods(ready, start)
+ waitForWatchReadiness(ready, resourceName)
+ getOutput, version, err = oc.subnetRegistry.GetPods()
default:
log.Fatalf("Unknown resource %s for watch and get resource", resourceName)
}
