Changes for caching pod ip (#600)
* Changes for cahing pod ip * Test fix for API changes * added test * Fixed merge conflicts * Add tests for pod cache * Add one more check to validate the cache * Incorporated the comment Co-authored-by: neaggarw <neaggarwMS@users.noreply.github.com>
This commit is contained in:
neaggarwMS
GitHub
Родитель
14e8a980cc
Коммит
508a2bb9c7
1
go.mod
1
go.mod
|
|
@ -12,7 +12,6 @@ require (
|
|||
github.com/docker/libnetwork v0.5.6
|
||||
github.com/golang/groupcache v0.0.0-20191027212112-611e8accdfc9 // indirect
|
||||
github.com/google/uuid v1.1.1
|
||||
github.com/googleapis/gnostic v0.3.1 // indirect
|
||||
github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 // indirect
|
||||
github.com/hashicorp/golang-lru v0.5.3 // indirect
|
||||
github.com/imdario/mergo v0.3.8 // indirect
|
||||
|
|
|
|||
|
|
@ -28,14 +28,15 @@ type IpsetManager struct {
|
|||
// Ipset represents one ipset entry.
|
||||
type Ipset struct {
|
||||
name string
|
||||
elements []string
|
||||
elements map[string]string // key = ip, value: context associated to the ip like podUid
|
||||
referCount int
|
||||
}
|
||||
|
||||
// NewIpset creates a new instance for Ipset object.
|
||||
func NewIpset(setName string) *Ipset {
|
||||
return &Ipset{
|
||||
name: setName,
|
||||
name: setName,
|
||||
elements: make(map[string]string),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -58,13 +59,11 @@ func (ipsMgr *IpsetManager) Exists(key string, val string, kind string) bool {
|
|||
return false
|
||||
}
|
||||
|
||||
for _, elem := range m[key].elements {
|
||||
if elem == val {
|
||||
return true
|
||||
}
|
||||
if _, exists := m[key].elements[val]; !exists {
|
||||
return false
|
||||
}
|
||||
|
||||
return false
|
||||
return true
|
||||
}
|
||||
|
||||
func isNsSet(setName string) bool {
|
||||
|
|
@ -140,7 +139,7 @@ func (ipsMgr *IpsetManager) AddToList(listName string, setName string) error {
|
|||
return err
|
||||
}
|
||||
|
||||
ipsMgr.listMap[listName].elements = append(ipsMgr.listMap[listName].elements, setName)
|
||||
ipsMgr.listMap[listName].elements[setName] = ""
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -152,12 +151,6 @@ func (ipsMgr *IpsetManager) DeleteFromList(listName string, setName string) erro
|
|||
return nil
|
||||
}
|
||||
|
||||
for i, val := range ipsMgr.listMap[listName].elements {
|
||||
if val == setName {
|
||||
ipsMgr.listMap[listName].elements = append(ipsMgr.listMap[listName].elements[:i], ipsMgr.listMap[listName].elements[i+1:]...)
|
||||
}
|
||||
}
|
||||
|
||||
hashedListName, hashedSetName := util.GetHashedName(listName), util.GetHashedName(setName)
|
||||
entry := &ipsEntry{
|
||||
operationFlag: util.IpsetDeletionFlag,
|
||||
|
|
@ -170,6 +163,11 @@ func (ipsMgr *IpsetManager) DeleteFromList(listName string, setName string) erro
|
|||
return err
|
||||
}
|
||||
|
||||
// Now cleanup the cache
|
||||
if _, exists := ipsMgr.listMap[listName].elements[setName]; exists {
|
||||
delete(ipsMgr.listMap[listName].elements, setName)
|
||||
}
|
||||
|
||||
if len(ipsMgr.listMap[listName].elements) == 0 {
|
||||
if err := ipsMgr.DeleteList(listName); err != nil {
|
||||
log.Errorf("Error: failed to delete ipset list %s.", listName)
|
||||
|
|
@ -231,8 +229,18 @@ func (ipsMgr *IpsetManager) DeleteSet(setName string) error {
|
|||
}
|
||||
|
||||
// AddToSet inserts an ip to an entry in setMap, and creates/updates the corresponding ipset.
|
||||
func (ipsMgr *IpsetManager) AddToSet(setName, ip, spec string) error {
|
||||
func (ipsMgr *IpsetManager) AddToSet(setName, ip, spec, podUid string) error {
|
||||
if ipsMgr.Exists(setName, ip, spec) {
|
||||
|
||||
// make sure we have updated the podUid in case it gets changed
|
||||
cachedPodUid := ipsMgr.setMap[setName].elements[ip]
|
||||
if cachedPodUid != podUid {
|
||||
log.Logf("AddToSet: PodOwner has changed for Ip: %s, setName:%s, Old podUid: %s, new PodUid: %s. Replace context with new PodOwner.",
|
||||
ip, setName, cachedPodUid, podUid)
|
||||
|
||||
ipsMgr.setMap[setName].elements[ip] = podUid
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -258,24 +266,32 @@ func (ipsMgr *IpsetManager) AddToSet(setName, ip, spec string) error {
|
|||
return err
|
||||
}
|
||||
|
||||
ipsMgr.setMap[setName].elements = append(ipsMgr.setMap[setName].elements, ip)
|
||||
// Stores the podUid as the context for this ip.
|
||||
ipsMgr.setMap[setName].elements[ip] = podUid
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteFromSet removes an ip from an entry in setMap, and delete/update the corresponding ipset.
|
||||
func (ipsMgr *IpsetManager) DeleteFromSet(setName, ip string) error {
|
||||
if _, exists := ipsMgr.setMap[setName]; !exists {
|
||||
func (ipsMgr *IpsetManager) DeleteFromSet(setName, ip, podUid string) error {
|
||||
ipSet, exists := ipsMgr.setMap[setName]
|
||||
if !exists {
|
||||
log.Logf("ipset with name %s not found", setName)
|
||||
return nil
|
||||
}
|
||||
|
||||
for i, val := range ipsMgr.setMap[setName].elements {
|
||||
if val == ip {
|
||||
ipsMgr.setMap[setName].elements = append(ipsMgr.setMap[setName].elements[:i], ipsMgr.setMap[setName].elements[i+1:]...)
|
||||
if _, exists := ipsMgr.setMap[setName].elements[ip]; exists {
|
||||
// in case the IP belongs to a new Pod, then ignore this Delete call as this might be stale
|
||||
cachedPodUid := ipSet.elements[ip]
|
||||
if cachedPodUid != podUid {
|
||||
log.Logf("DeleteFromSet: PodOwner has changed for Ip: %s, setName:%s, Old podUid: %s, new PodUid: %s. Ignore the delete as this is stale update",
|
||||
ip, setName, cachedPodUid, podUid)
|
||||
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// TODO optimize to not run this command in case cache has already been updated.
|
||||
entry := &ipsEntry{
|
||||
operationFlag: util.IpsetDeletionFlag,
|
||||
set: util.GetHashedName(setName),
|
||||
|
|
@ -291,6 +307,9 @@ func (ipsMgr *IpsetManager) DeleteFromSet(setName, ip string) error {
|
|||
return err
|
||||
}
|
||||
|
||||
// Now cleanup the cache
|
||||
delete(ipsMgr.setMap[setName].elements, ip)
|
||||
|
||||
if len(ipsMgr.setMap[setName].elements) == 0 {
|
||||
ipsMgr.DeleteSet(setName)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -170,15 +170,55 @@ func TestAddToSet(t *testing.T) {
|
|||
}
|
||||
}()
|
||||
|
||||
if err := ipsMgr.AddToSet("test-set", "1.2.3.4", util.IpsetNetHashFlag); err != nil {
|
||||
if err := ipsMgr.AddToSet("test-set", "1.2.3.4", util.IpsetNetHashFlag, ""); err != nil {
|
||||
t.Errorf("TestAddToSet failed @ ipsMgr.AddToSet")
|
||||
}
|
||||
|
||||
if err := ipsMgr.AddToSet("test-set", "1.2.3.4/nomatch", util.IpsetNetHashFlag); err != nil {
|
||||
if err := ipsMgr.AddToSet("test-set", "1.2.3.4/nomatch", util.IpsetNetHashFlag, ""); err != nil {
|
||||
t.Errorf("TestAddToSet with nomatch failed @ ipsMgr.AddToSet")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAddToSetWithCachePodInfo(t *testing.T) {
|
||||
ipsMgr := NewIpsetManager()
|
||||
if err := ipsMgr.Save(util.IpsetTestConfigFile); err != nil {
|
||||
t.Errorf("TestAddToSetWithCachePodInfo failed @ ipsMgr.Save")
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if err := ipsMgr.Restore(util.IpsetTestConfigFile); err != nil {
|
||||
t.Errorf("TestAddToSetWithCachePodInfo failed @ ipsMgr.Restore")
|
||||
}
|
||||
}()
|
||||
|
||||
var pod1 = "pod1"
|
||||
var setname = "test-podcache_new"
|
||||
var ip = "10.0.2.7"
|
||||
if err := ipsMgr.AddToSet(setname, ip, util.IpsetNetHashFlag, pod1); err != nil {
|
||||
t.Errorf("TestAddToSetWithCachePodInfo with pod1 failed @ ipsMgr.AddToSet, setname: %s, hashedname: %s", setname, util.GetHashedName(setname))
|
||||
}
|
||||
|
||||
// validate if Pod1 exists
|
||||
cachedPodUid := ipsMgr.setMap[setname].elements[ip]
|
||||
if cachedPodUid != pod1 {
|
||||
t.Errorf("setname: %s, hashedname: %s is added with wrong podUid: %s, expected: %s", setname, util.GetHashedName(setname), cachedPodUid, pod1)
|
||||
}
|
||||
|
||||
// now add pod2 with the same ip. This is possible if DeletePod1 is handled after AddPod2 event callback.
|
||||
var pod2 = "pod2"
|
||||
if err := ipsMgr.AddToSet(setname, ip, util.IpsetNetHashFlag, pod2); err != nil {
|
||||
t.Errorf("TestAddToSetWithCachePodInfo with pod2 failed @ ipsMgr.AddToSet")
|
||||
}
|
||||
|
||||
cachedPodUid = ipsMgr.setMap[setname].elements[ip]
|
||||
if cachedPodUid != pod2 {
|
||||
t.Errorf("setname: %s, hashedname: %s is added with wrong podUid: %s, expected: %s", setname, util.GetHashedName(setname), cachedPodUid, pod2)
|
||||
}
|
||||
|
||||
// Delete from set, it will delete the set if this is the last member
|
||||
ipsMgr.DeleteFromSet(setname, ip, pod2)
|
||||
}
|
||||
|
||||
func TestDeleteFromSet(t *testing.T) {
|
||||
ipsMgr := NewIpsetManager()
|
||||
if err := ipsMgr.Save(util.IpsetTestConfigFile); err != nil {
|
||||
|
|
@ -191,7 +231,7 @@ func TestDeleteFromSet(t *testing.T) {
|
|||
}
|
||||
}()
|
||||
|
||||
if err := ipsMgr.AddToSet("test-set", "1.2.3.4", util.IpsetNetHashFlag); err != nil {
|
||||
if err := ipsMgr.AddToSet("test-set", "1.2.3.4", util.IpsetNetHashFlag, ""); err != nil {
|
||||
t.Errorf("TestDeleteFromSet failed @ ipsMgr.AddToSet")
|
||||
}
|
||||
|
||||
|
|
@ -199,7 +239,7 @@ func TestDeleteFromSet(t *testing.T) {
|
|||
t.Errorf("TestDeleteFromSet failed @ ipsMgr.AddToSet")
|
||||
}
|
||||
|
||||
if err := ipsMgr.DeleteFromSet("test-set", "1.2.3.4"); err != nil {
|
||||
if err := ipsMgr.DeleteFromSet("test-set", "1.2.3.4", ""); err != nil {
|
||||
t.Errorf("TestDeleteFromSet failed @ ipsMgr.DeleteFromSet")
|
||||
}
|
||||
|
||||
|
|
@ -209,6 +249,65 @@ func TestDeleteFromSet(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestDeleteFromSetWithPodCache(t *testing.T) {
|
||||
ipsMgr := NewIpsetManager()
|
||||
if err := ipsMgr.Save(util.IpsetTestConfigFile); err != nil {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache failed @ ipsMgr.Save")
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if err := ipsMgr.Restore(util.IpsetTestConfigFile); err != nil {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache failed @ ipsMgr.Restore")
|
||||
}
|
||||
}()
|
||||
|
||||
var setname = "test-deleteset-withcache"
|
||||
var ip = "10.0.2.8"
|
||||
var pod1 = "pod1"
|
||||
if err := ipsMgr.AddToSet(setname, ip, util.IpsetNetHashFlag, pod1); err != nil {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache failed for pod1 @ ipsMgr.AddToSet")
|
||||
}
|
||||
|
||||
if len(ipsMgr.setMap[setname].elements) != 1 {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache failed @ ipsMgr.AddToSet")
|
||||
}
|
||||
|
||||
if err := ipsMgr.DeleteFromSet(setname, ip, pod1); err != nil {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache for pod1 failed @ ipsMgr.DeleteFromSet")
|
||||
}
|
||||
|
||||
// now add the set again and then replace it with pod2
|
||||
var pod2 = "pod2"
|
||||
if err := ipsMgr.AddToSet(setname, ip, util.IpsetNetHashFlag, pod1); err != nil {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache failed for pod1 @ ipsMgr.AddToSet")
|
||||
}
|
||||
|
||||
// Add Pod2 with same ip (This could happen if AddPod2 is served before DeletePod1)
|
||||
if err := ipsMgr.AddToSet(setname, ip, util.IpsetNetHashFlag, pod2); err != nil {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache failed for pod2 @ ipsMgr.AddToSet")
|
||||
}
|
||||
|
||||
// Process DeletePod1
|
||||
if err := ipsMgr.DeleteFromSet(setname, ip, pod1); err != nil {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache for pod1 failed @ ipsMgr.DeleteFromSet")
|
||||
}
|
||||
|
||||
// note the set will stil exist with pod ip
|
||||
cachedPodUid := ipsMgr.setMap[setname].elements[ip]
|
||||
if cachedPodUid != pod2 {
|
||||
t.Errorf("setname: %s, hashedname: %s is added with wrong podUid: %s, expected: %s", setname, util.GetHashedName(setname), cachedPodUid, pod2)
|
||||
}
|
||||
|
||||
// Now cleanup and delete pod2
|
||||
if err := ipsMgr.DeleteFromSet(setname, ip, pod2); err != nil {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache for pod2 failed @ ipsMgr.DeleteFromSet")
|
||||
}
|
||||
|
||||
if _, exists := ipsMgr.setMap[setname]; exists {
|
||||
t.Errorf("TestDeleteFromSetWithPodCache failed @ ipsMgr.DeleteFromSet")
|
||||
}
|
||||
}
|
||||
|
||||
func TestClean(t *testing.T) {
|
||||
ipsMgr := NewIpsetManager()
|
||||
if err := ipsMgr.Save(util.IpsetTestConfigFile); err != nil {
|
||||
|
|
@ -242,7 +341,7 @@ func TestDestroy(t *testing.T) {
|
|||
}
|
||||
}()
|
||||
|
||||
if err := ipsMgr.AddToSet("test-set", "1.2.3.4", util.IpsetNetHashFlag); err != nil {
|
||||
if err := ipsMgr.AddToSet("test-set", "1.2.3.4", util.IpsetNetHashFlag, ""); err != nil {
|
||||
t.Errorf("TestDestroy failed @ ipsMgr.AddToSet")
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -47,7 +47,7 @@ type NetworkPolicyManager struct {
|
|||
|
||||
nodeName string
|
||||
nsMap map[string]*namespace
|
||||
podMap map[string]bool
|
||||
podMap map[string]string // Key: Pod uuid, Value: PodIp
|
||||
isAzureNpmChainCreated bool
|
||||
isSafeToCleanUpAzureNpmChain bool
|
||||
|
||||
|
|
@ -234,7 +234,7 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
|
|||
npInformer: npInformer,
|
||||
nodeName: os.Getenv("HOSTNAME"),
|
||||
nsMap: make(map[string]*namespace),
|
||||
podMap: make(map[string]bool),
|
||||
podMap: make(map[string]string),
|
||||
isAzureNpmChainCreated: false,
|
||||
isSafeToCleanUpAzureNpmChain: false,
|
||||
clusterState: telemetry.ClusterState{
|
||||
|
|
|
|||
|
|
@ -6,8 +6,8 @@ import (
|
|||
"strconv"
|
||||
|
||||
"github.com/Azure/azure-container-networking/log"
|
||||
"github.com/Azure/azure-container-networking/npm/iptm"
|
||||
"github.com/Azure/azure-container-networking/npm/ipsm"
|
||||
"github.com/Azure/azure-container-networking/npm/iptm"
|
||||
"github.com/Azure/azure-container-networking/npm/util"
|
||||
networkingv1 "k8s.io/api/networking/v1"
|
||||
)
|
||||
|
|
@ -66,12 +66,12 @@ func (npMgr *NetworkPolicyManager) AddNetworkPolicy(npObj *networkingv1.NetworkP
|
|||
}
|
||||
|
||||
var (
|
||||
hashedSelector = HashSelector(&npObj.Spec.PodSelector)
|
||||
addedPolicy *networkingv1.NetworkPolicy
|
||||
sets, namedPorts, lists []string
|
||||
ingressIPCidrs, egressIPCidrs [][]string
|
||||
iptEntries []*iptm.IptEntry
|
||||
ipsMgr = allNs.ipsMgr
|
||||
hashedSelector = HashSelector(&npObj.Spec.PodSelector)
|
||||
addedPolicy *networkingv1.NetworkPolicy
|
||||
sets, namedPorts, lists []string
|
||||
ingressIPCidrs, egressIPCidrs [][]string
|
||||
iptEntries []*iptm.IptEntry
|
||||
ipsMgr = allNs.ipsMgr
|
||||
)
|
||||
|
||||
// Remove the existing policy from processed (merged) network policy map
|
||||
|
|
@ -211,17 +211,17 @@ func createCidrsRule(ingressOrEgress, policyName, ns string, ipsetEntries [][]st
|
|||
log.Printf("Error creating ipset %s", ipCidrSet)
|
||||
}
|
||||
for _, ipCidrEntry := range util.DropEmptyFields(ipCidrSet) {
|
||||
// Ipset doesn't allow 0.0.0.0/0 to be added. A general solution is split 0.0.0.0/1 in half which convert to
|
||||
// Ipset doesn't allow 0.0.0.0/0 to be added. A general solution is split 0.0.0.0/1 in half which convert to
|
||||
// 1.0.0.0/1 and 128.0.0.0/1
|
||||
if (ipCidrEntry == "0.0.0.0/0") {
|
||||
if ipCidrEntry == "0.0.0.0/0" {
|
||||
splitEntry := [2]string{"1.0.0.0/1", "128.0.0.0/1"}
|
||||
for _, entry := range splitEntry {
|
||||
if err := ipsMgr.AddToSet(setName, entry, util.IpsetNetHashFlag); err != nil {
|
||||
if err := ipsMgr.AddToSet(setName, entry, util.IpsetNetHashFlag, ""); err != nil {
|
||||
log.Printf("Error adding ip cidrs %s into ipset %s", entry, ipCidrSet)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if err := ipsMgr.AddToSet(setName, ipCidrEntry, util.IpsetNetHashFlag); err != nil {
|
||||
if err := ipsMgr.AddToSet(setName, ipCidrEntry, util.IpsetNetHashFlag, ""); err != nil {
|
||||
log.Printf("Error adding ip cidrs %s into ipset %s", ipCidrEntry, ipCidrSet)
|
||||
}
|
||||
}
|
||||
|
|
@ -240,4 +240,4 @@ func removeCidrsRule(ingressOrEgress, policyName, ns string, ipsetEntries [][]st
|
|||
log.Printf("Error deleting ipset %s", ipCidrSet)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
50
npm/pod.go
50
npm/pod.go
|
|
@ -42,6 +42,7 @@ func (npMgr *NetworkPolicyManager) AddPod(podObj *corev1.Pod) error {
|
|||
var (
|
||||
err error
|
||||
podNs = "ns-" + podObj.ObjectMeta.Namespace
|
||||
podUid = string(podObj.ObjectMeta.UID)
|
||||
podName = podObj.ObjectMeta.Name
|
||||
podNodeName = podObj.Spec.NodeName
|
||||
podLabels = podObj.ObjectMeta.Labels
|
||||
|
|
@ -50,7 +51,7 @@ func (npMgr *NetworkPolicyManager) AddPod(podObj *corev1.Pod) error {
|
|||
ipsMgr = npMgr.nsMap[util.KubeAllNamespacesFlag].ipsMgr
|
||||
)
|
||||
|
||||
log.Logf("POD CREATING: [%s/%s/%s%+v%s]", podNs, podName, podNodeName, podLabels, podIP)
|
||||
log.Logf("POD CREATING: [%s%s/%s/%s%+v%s]", podUid, podNs, podName, podNodeName, podLabels, podIP)
|
||||
|
||||
// Add pod namespace if it doesn't exist
|
||||
if _, exists := npMgr.nsMap[podNs]; !exists {
|
||||
|
|
@ -62,20 +63,20 @@ func (npMgr *NetworkPolicyManager) AddPod(podObj *corev1.Pod) error {
|
|||
|
||||
// Add the pod to its namespace's ipset.
|
||||
log.Logf("Adding pod %s to ipset %s", podIP, podNs)
|
||||
if err = ipsMgr.AddToSet(podNs, podIP, util.IpsetNetHashFlag); err != nil {
|
||||
if err = ipsMgr.AddToSet(podNs, podIP, util.IpsetNetHashFlag, podUid); err != nil {
|
||||
log.Errorf("Error: failed to add pod to namespace ipset.")
|
||||
}
|
||||
|
||||
// Add the pod to its label's ipset.
|
||||
for podLabelKey, podLabelVal := range podLabels {
|
||||
log.Logf("Adding pod %s to ipset %s", podIP, podLabelKey)
|
||||
if err = ipsMgr.AddToSet(podLabelKey, podIP, util.IpsetNetHashFlag); err != nil {
|
||||
if err = ipsMgr.AddToSet(podLabelKey, podIP, util.IpsetNetHashFlag, podUid); err != nil {
|
||||
log.Errorf("Error: failed to add pod to label ipset.")
|
||||
}
|
||||
|
||||
label := podLabelKey + ":" + podLabelVal
|
||||
log.Logf("Adding pod %s to ipset %s", podIP, label)
|
||||
if err = ipsMgr.AddToSet(label, podIP, util.IpsetNetHashFlag); err != nil {
|
||||
if err = ipsMgr.AddToSet(label, podIP, util.IpsetNetHashFlag, podUid); err != nil {
|
||||
log.Errorf("Error: failed to add pod to label ipset.")
|
||||
}
|
||||
}
|
||||
|
|
@ -91,12 +92,13 @@ func (npMgr *NetworkPolicyManager) AddPod(podObj *corev1.Pod) error {
|
|||
case v1.ProtocolSCTP:
|
||||
protocol = util.IpsetSCTPFlag
|
||||
}
|
||||
ipsMgr.AddToSet(port.Name, fmt.Sprintf("%s,%s%d", podIP, protocol, port.ContainerPort), util.IpsetIPPortHashFlag)
|
||||
ipsMgr.AddToSet(port.Name, fmt.Sprintf("%s,%s%d", podIP, protocol, port.ContainerPort), util.IpsetIPPortHashFlag, podUid)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
npMgr.podMap[podNs+podName] = true
|
||||
// add the Pod info to the podMap
|
||||
npMgr.podMap[podUid] = podIP
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -131,6 +133,8 @@ func (npMgr *NetworkPolicyManager) UpdatePod(oldPodObj, newPodObj *corev1.Pod) e
|
|||
newPodObjNs, newPodObjName, newPodObjLabel, newPodObjPhase, newPodObjIP,
|
||||
)
|
||||
|
||||
// Todo: Update if cached ip and podip changed and it is not a delete event
|
||||
|
||||
if err = npMgr.DeletePod(oldPodObj); err != nil {
|
||||
log.Errorf("Error: failed to delete pod during update with error %+v", err)
|
||||
return err
|
||||
|
|
@ -138,7 +142,8 @@ func (npMgr *NetworkPolicyManager) UpdatePod(oldPodObj, newPodObj *corev1.Pod) e
|
|||
|
||||
// Assume that the pod IP will be released when pod moves to succeeded or failed state.
|
||||
// If the pod transitions back to an active state, then add operation will re establish the updated pod info.
|
||||
if newPodObj.ObjectMeta.DeletionTimestamp == nil && newPodObj.ObjectMeta.DeletionGracePeriodSeconds == nil && newPodObjPhase != v1.PodSucceeded && newPodObjPhase != v1.PodFailed {
|
||||
if newPodObj.ObjectMeta.DeletionTimestamp == nil && newPodObj.ObjectMeta.DeletionGracePeriodSeconds == nil &&
|
||||
newPodObjPhase != v1.PodSucceeded && newPodObjPhase != v1.PodFailed {
|
||||
if err = npMgr.AddPod(newPodObj); err != nil {
|
||||
log.Errorf("Error: failed to add pod during update with error %+v", err)
|
||||
}
|
||||
|
|
@ -149,43 +154,46 @@ func (npMgr *NetworkPolicyManager) UpdatePod(oldPodObj, newPodObj *corev1.Pod) e
|
|||
|
||||
// DeletePod handles deleting pod from its label's ipset.
|
||||
func (npMgr *NetworkPolicyManager) DeletePod(podObj *corev1.Pod) error {
|
||||
if !isValidPod(podObj) {
|
||||
return nil
|
||||
}
|
||||
|
||||
var (
|
||||
err error
|
||||
podNs = "ns-" + podObj.ObjectMeta.Namespace
|
||||
podUid = string(podObj.ObjectMeta.UID)
|
||||
podName = podObj.ObjectMeta.Name
|
||||
podNodeName = podObj.Spec.NodeName
|
||||
podLabels = podObj.ObjectMeta.Labels
|
||||
podIP = podObj.Status.PodIP
|
||||
podContainers = podObj.Spec.Containers
|
||||
ipsMgr = npMgr.nsMap[util.KubeAllNamespacesFlag].ipsMgr
|
||||
)
|
||||
|
||||
_, exists := npMgr.podMap[podNs+podName]
|
||||
cachedPodIp, exists := npMgr.podMap[podUid]
|
||||
if !exists {
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Logf("POD DELETING: [%s/%s/%s%+v%s]", podNs, podName, podNodeName, podLabels, podIP)
|
||||
// if the podIp exists, it must match the cachedIp
|
||||
if len(podObj.Status.PodIP) > 0 && cachedPodIp != podObj.Status.PodIP {
|
||||
// TODO Add AI telemetry event
|
||||
log.Errorf("Error: Unexpected state. Pod (Namespace:%s, Name:%s, uid:%s, has cachedPodIp:%s which is different from PodIp:%s",
|
||||
podNs, podName, podUid, cachedPodIp, podObj.Status.PodIP)
|
||||
}
|
||||
|
||||
log.Logf("POD DELETING: [%s/%s%s/%s%+v%s]", podNs, podName, podUid, podNodeName, podLabels, cachedPodIp)
|
||||
|
||||
// Delete the pod from its namespace's ipset.
|
||||
if err = ipsMgr.DeleteFromSet(podNs, podIP); err != nil {
|
||||
if err = ipsMgr.DeleteFromSet(podNs, cachedPodIp, podUid); err != nil {
|
||||
log.Errorf("Error: failed to delete pod from namespace ipset.")
|
||||
}
|
||||
|
||||
// Delete the pod from its label's ipset.
|
||||
for podLabelKey, podLabelVal := range podLabels {
|
||||
log.Logf("Deleting pod %s from ipset %s", podIP, podLabelKey)
|
||||
if err = ipsMgr.DeleteFromSet(podLabelKey, podIP); err != nil {
|
||||
log.Logf("Deleting pod %s from ipset %s", cachedPodIp, podLabelKey)
|
||||
if err = ipsMgr.DeleteFromSet(podLabelKey, cachedPodIp, podUid); err != nil {
|
||||
log.Errorf("Error: failed to delete pod from label ipset.")
|
||||
}
|
||||
|
||||
label := podLabelKey + ":" + podLabelVal
|
||||
log.Logf("Deleting pod %s from ipset %s", podIP, label)
|
||||
if err = ipsMgr.DeleteFromSet(label, podIP); err != nil {
|
||||
log.Logf("Deleting pod %s from ipset %s", cachedPodIp, label)
|
||||
if err = ipsMgr.DeleteFromSet(label, cachedPodIp, podUid); err != nil {
|
||||
log.Errorf("Error: failed to delete pod from label ipset.")
|
||||
}
|
||||
}
|
||||
|
|
@ -201,12 +209,12 @@ func (npMgr *NetworkPolicyManager) DeletePod(podObj *corev1.Pod) error {
|
|||
case v1.ProtocolSCTP:
|
||||
protocol = util.IpsetSCTPFlag
|
||||
}
|
||||
ipsMgr.DeleteFromSet(port.Name, fmt.Sprintf("%s,%s%d", podIP, protocol, port.ContainerPort))
|
||||
ipsMgr.DeleteFromSet(port.Name, fmt.Sprintf("%s,%s%d", cachedPodIp, protocol, port.ContainerPort), podUid)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
delete(npMgr.podMap, podNs+podName)
|
||||
delete(npMgr.podMap, podUid)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ func TestisSystemPod(t *testing.T) {
|
|||
func TestAddPod(t *testing.T) {
|
||||
npMgr := &NetworkPolicyManager{
|
||||
nsMap: make(map[string]*namespace),
|
||||
podMap: make(map[string]bool),
|
||||
podMap: make(map[string]string),
|
||||
TelemetryEnabled: false,
|
||||
}
|
||||
|
||||
|
|
@ -82,7 +82,7 @@ func TestAddPod(t *testing.T) {
|
|||
func TestUpdatePod(t *testing.T) {
|
||||
npMgr := &NetworkPolicyManager{
|
||||
nsMap: make(map[string]*namespace),
|
||||
podMap: make(map[string]bool),
|
||||
podMap: make(map[string]string),
|
||||
TelemetryEnabled: false,
|
||||
}
|
||||
|
||||
|
|
@ -145,7 +145,7 @@ func TestUpdatePod(t *testing.T) {
|
|||
func TestDeletePod(t *testing.T) {
|
||||
npMgr := &NetworkPolicyManager{
|
||||
nsMap: make(map[string]*namespace),
|
||||
podMap: make(map[string]bool),
|
||||
podMap: make(map[string]string),
|
||||
TelemetryEnabled: false,
|
||||
}
|
||||
|
||||
|
|
|
|||
Загрузка…
Ссылка в новой задаче