Skip to content

Lock update operations to prevent race in between updates. #536

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Apr 3, 2020
Merged

Lock update operations to prevent race in between updates. #536

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1fcfb0d
Lock update operations to prevent race in between updates.
jaer-tsun Apr 2, 2020
08477c5
fixing tests
jaer-tsun Apr 2, 2020
3383b07
fixing nwpolicy files
jaer-tsun Apr 3, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 5 additions & 11 deletions npm/namespace.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ func (ns *namespace) policyExists(npObj *networkingv1.NetworkPolicy) bool {
// InitAllNsList syncs all-namespace ipset list.
func (npMgr *NetworkPolicyManager) InitAllNsList() error {
allNs := npMgr.nsMap[util.KubeAllNamespacesFlag]
for ns:= range npMgr.nsMap {
for ns := range npMgr.nsMap {
if ns == util.KubeAllNamespacesFlag {
continue
}
Expand Down Expand Up @@ -88,12 +88,9 @@ func (npMgr *NetworkPolicyManager) UninitAllNsList() error {

// AddNamespace handles adding namespace to ipset.
func (npMgr *NetworkPolicyManager) AddNamespace(nsObj *corev1.Namespace) error {
npMgr.Lock()
defer npMgr.Unlock()

var err error

nsName, nsLabel := "ns-" + nsObj.ObjectMeta.Name, nsObj.ObjectMeta.Labels
nsName, nsLabel := "ns-"+nsObj.ObjectMeta.Name, nsObj.ObjectMeta.Labels
log.Printf("NAMESPACE CREATING: [%s/%v]", nsName, nsLabel)

ipsMgr := npMgr.nsMap[util.KubeAllNamespacesFlag].ipsMgr
Expand Down Expand Up @@ -139,8 +136,8 @@ func (npMgr *NetworkPolicyManager) AddNamespace(nsObj *corev1.Namespace) error {
func (npMgr *NetworkPolicyManager) UpdateNamespace(oldNsObj *corev1.Namespace, newNsObj *corev1.Namespace) error {
var err error

oldNsNs, oldNsLabel := "ns-" + oldNsObj.ObjectMeta.Name, oldNsObj.ObjectMeta.Labels
newNsNs, newNsLabel := "ns-" + newNsObj.ObjectMeta.Name, newNsObj.ObjectMeta.Labels
oldNsNs, oldNsLabel := "ns-"+oldNsObj.ObjectMeta.Name, oldNsObj.ObjectMeta.Labels
newNsNs, newNsLabel := "ns-"+newNsObj.ObjectMeta.Name, newNsObj.ObjectMeta.Labels
log.Printf(
"NAMESPACE UPDATING:\n old namespace: [%s/%v]\n new namespace: [%s/%v]",
oldNsNs, oldNsLabel, newNsNs, newNsLabel,
Expand All @@ -161,12 +158,9 @@ func (npMgr *NetworkPolicyManager) UpdateNamespace(oldNsObj *corev1.Namespace, n

// DeleteNamespace handles deleting namespace from ipset.
func (npMgr *NetworkPolicyManager) DeleteNamespace(nsObj *corev1.Namespace) error {
npMgr.Lock()
defer npMgr.Unlock()

var err error

nsName, nsLabel := "ns-" + nsObj.ObjectMeta.Name, nsObj.ObjectMeta.Labels
nsName, nsLabel := "ns-"+nsObj.ObjectMeta.Name, nsObj.ObjectMeta.Labels
log.Printf("NAMESPACE DELETING: [%s/%v]", nsName, nsLabel)

_, exists := npMgr.nsMap[nsName]
Expand Down
8 changes: 7 additions & 1 deletion npm/namespace_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,8 @@
package npm

import (
"testing"
"os"
"testing"

"github.com/Azure/azure-container-networking/npm/iptm"

Expand Down Expand Up @@ -75,9 +75,11 @@ func TestAddNamespace(t *testing.T) {
},
}

npMgr.Lock()
if err := npMgr.AddNamespace(nsObj); err != nil {
t.Errorf("TestAddNamespace @ npMgr.AddNamespace")
}
npMgr.Unlock()
}

func TestUpdateNamespace(t *testing.T) {
Expand Down Expand Up @@ -121,13 +123,15 @@ func TestUpdateNamespace(t *testing.T) {
},
}

npMgr.Lock()
if err := npMgr.AddNamespace(oldNsObj); err != nil {
t.Errorf("TestUpdateNamespace failed @ npMgr.AddNamespace")
}

if err := npMgr.UpdateNamespace(oldNsObj, newNsObj); err != nil {
t.Errorf("TestUpdateNamespace failed @ npMgr.UpdateNamespace")
}
npMgr.Unlock()
}

func TestDeleteNamespace(t *testing.T) {
Expand Down Expand Up @@ -162,13 +166,15 @@ func TestDeleteNamespace(t *testing.T) {
},
}

npMgr.Lock()
if err := npMgr.AddNamespace(nsObj); err != nil {
t.Errorf("TestDeleteNamespace @ npMgr.AddNamespace")
}

if err := npMgr.DeleteNamespace(nsObj); err != nil {
t.Errorf("TestDeleteNamespace @ npMgr.DeleteNamespace")
}
npMgr.Unlock()
}

func TestMain(m *testing.M) {
Expand Down
44 changes: 31 additions & 13 deletions npm/npm.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,8 @@ type NetworkPolicyManager struct {
isAzureNpmChainCreated bool
isSafeToCleanUpAzureNpmChain bool

clusterState telemetry.ClusterState
version string
clusterState telemetry.ClusterState
version string

serverVersion *version.Info
TelemetryEnabled bool
Expand Down Expand Up @@ -93,26 +93,26 @@ func (npMgr *NetworkPolicyManager) SendAiMetrics() {
GetEnvRetryCount: 5,
GetEnvRetryWaitTimeInSecs: 3,
}
th, err = aitelemetry.NewAITelemetry("", aiMetadata, aiConfig)
heartbeat = time.NewTicker(time.Minute * heartbeatIntervalInMinutes).C

th, err = aitelemetry.NewAITelemetry("", aiMetadata, aiConfig)
heartbeat = time.NewTicker(time.Minute * heartbeatIntervalInMinutes).C
customDimensions = map[string]string{"ClusterID": util.GetClusterID(npMgr.nodeName),
"APIServer": npMgr.serverVersion.String()}
"APIServer": npMgr.serverVersion.String()}
podCount = aitelemetry.Metric{
Name: "PodCount",
Name: "PodCount",
CustomDimensions: customDimensions,
}
nsCount = aitelemetry.Metric{
Name: "NsCount",
Name: "NsCount",
CustomDimensions: customDimensions,
}
nwPolicyCount = aitelemetry.Metric{
Name: "NwPolicyCount",
Name: "NwPolicyCount",
CustomDimensions: customDimensions,
}
)

for i := 0; err != nil && i < 5; i++{
for i := 0; err != nil && i < 5; i++ {
log.Logf("Failed to init AppInsights with err: %+v", err)
time.Sleep(time.Minute * 5)
th, err = aitelemetry.NewAITelemetry("", aiMetadata, aiConfig)
Expand All @@ -123,7 +123,7 @@ func (npMgr *NetworkPolicyManager) SendAiMetrics() {

defer th.Close(10)

for {
for {
<-heartbeat
clusterState := npMgr.GetClusterState()
podCount.Value = float64(clusterState.PodCount)
Expand Down Expand Up @@ -206,7 +206,7 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
err error
)

for ticker, start := time.NewTicker(1 * time.Second).C, time.Now(); time.Since(start) < time.Minute * 1; {
for ticker, start := time.NewTicker(1*time.Second).C, time.Now(); time.Since(start) < time.Minute*1; {
<-ticker
serverVersion, err = clientset.ServerVersion()
if err == nil {
Expand Down Expand Up @@ -239,7 +239,7 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
NsCount: 0,
NwPolicyCount: 0,
},
version: npmVersion,
version: npmVersion,
serverVersion: serverVersion,
TelemetryEnabled: true,
}
Expand All @@ -257,13 +257,19 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
// Pod event handlers
cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
npMgr.Lock()
npMgr.AddPod(obj.(*corev1.Pod))
npMgr.Unlock()
},
UpdateFunc: func(old, new interface{}) {
npMgr.Lock()
npMgr.UpdatePod(old.(*corev1.Pod), new.(*corev1.Pod))
npMgr.Unlock()
},
DeleteFunc: func(obj interface{}) {
npMgr.Lock()
npMgr.DeletePod(obj.(*corev1.Pod))
npMgr.Unlock()
},
},
)
Expand All @@ -272,13 +278,19 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
// Namespace event handlers
cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
npMgr.Lock()
npMgr.AddNamespace(obj.(*corev1.Namespace))
npMgr.Unlock()
},
UpdateFunc: func(old, new interface{}) {
npMgr.Lock()
npMgr.UpdateNamespace(old.(*corev1.Namespace), new.(*corev1.Namespace))
npMgr.Unlock()
},
DeleteFunc: func(obj interface{}) {
npMgr.Lock()
npMgr.DeleteNamespace(obj.(*corev1.Namespace))
npMgr.Unlock()
},
},
)
Expand All @@ -287,13 +299,19 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
// Network policy event handlers
cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
npMgr.Lock()
npMgr.AddNetworkPolicy(obj.(*networkingv1.NetworkPolicy))
npMgr.Unlock()
},
UpdateFunc: func(old, new interface{}) {
npMgr.Lock()
npMgr.UpdateNetworkPolicy(old.(*networkingv1.NetworkPolicy), new.(*networkingv1.NetworkPolicy))
npMgr.Unlock()
},
DeleteFunc: func(obj interface{}) {
npMgr.Lock()
npMgr.DeleteNetworkPolicy(obj.(*networkingv1.NetworkPolicy))
npMgr.Unlock()
},
},
)
Expand Down
8 changes: 0 additions & 8 deletions npm/nwpolicy.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,9 +25,6 @@ func (npMgr *NetworkPolicyManager) canCleanUpNpmChains() bool {

// AddNetworkPolicy handles adding network policy to iptables.
func (npMgr *NetworkPolicyManager) AddNetworkPolicy(npObj *networkingv1.NetworkPolicy) error {
npMgr.Lock()
defer npMgr.Unlock()

var (
err error
ns *namespace
Expand Down Expand Up @@ -75,9 +72,7 @@ func (npMgr *NetworkPolicyManager) AddNetworkPolicy(npObj *networkingv1.NetworkP
log.Printf("Error adding policy %s to %s", npName, oldPolicy.ObjectMeta.Name)
}
npMgr.isSafeToCleanUpAzureNpmChain = false
npMgr.Unlock()
npMgr.DeleteNetworkPolicy(oldPolicy)
npMgr.Lock()
npMgr.isSafeToCleanUpAzureNpmChain = true
} else {
ns.processedNpMap[hashedSelector] = npObj
Expand Down Expand Up @@ -140,9 +135,6 @@ func (npMgr *NetworkPolicyManager) UpdateNetworkPolicy(oldNpObj *networkingv1.Ne

// DeleteNetworkPolicy handles deleting network policy from iptables.
func (npMgr *NetworkPolicyManager) DeleteNetworkPolicy(npObj *networkingv1.NetworkPolicy) error {
npMgr.Lock()
defer npMgr.Unlock()

var (
err error
ns *namespace
Expand Down
18 changes: 16 additions & 2 deletions npm/nwpolicy_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,9 +61,11 @@ func TestAddNetworkPolicy(t *testing.T) {
},
}

npMgr.Lock()
if err := npMgr.AddNamespace(nsObj); err != nil {
t.Errorf("TestAddNetworkPolicy @ npMgr.AddNamespace")
}
npMgr.Unlock()

tcp := corev1.ProtocolTCP
port8000 := intstr.FromInt(8000)
Expand All @@ -82,17 +84,19 @@ func TestAddNetworkPolicy(t *testing.T) {
}},
Ports: []networkingv1.NetworkPolicyPort{{
Protocol: &tcp,
Port: &port8000,
Port: &port8000,
}},
},
},
},
}

npMgr.Lock()
if err := npMgr.AddNetworkPolicy(allowIngress); err != nil {
t.Errorf("TestAddNetworkPolicy failed @ allowIngress AddNetworkPolicy")
t.Errorf("Error: %v", err)
}
npMgr.Unlock()

allowEgress := &networkingv1.NetworkPolicy{
ObjectMeta: metav1.ObjectMeta{
Expand All @@ -109,17 +113,19 @@ func TestAddNetworkPolicy(t *testing.T) {
}},
Ports: []networkingv1.NetworkPolicyPort{{
Protocol: &tcp,
Port: &port8000,
Port: &port8000,
}},
},
},
},
}

npMgr.Lock()
if err := npMgr.AddNetworkPolicy(allowEgress); err != nil {
t.Errorf("TestAddNetworkPolicy failed @ allowEgress AddNetworkPolicy")
t.Errorf("Error: %v", err)
}
npMgr.Unlock()
}

func TestUpdateNetworkPolicy(t *testing.T) {
Expand Down Expand Up @@ -168,9 +174,11 @@ func TestUpdateNetworkPolicy(t *testing.T) {
},
}

npMgr.Lock()
if err := npMgr.AddNamespace(nsObj); err != nil {
t.Errorf("TestUpdateNetworkPolicy @ npMgr.AddNamespace")
}
npMgr.Unlock()

tcp, udp := corev1.ProtocolTCP, corev1.ProtocolUDP
allowIngress := &networkingv1.NetworkPolicy{
Expand Down Expand Up @@ -221,13 +229,15 @@ func TestUpdateNetworkPolicy(t *testing.T) {
},
}

npMgr.Lock()
if err := npMgr.AddNetworkPolicy(allowIngress); err != nil {
t.Errorf("TestUpdateNetworkPolicy failed @ AddNetworkPolicy")
}

if err := npMgr.UpdateNetworkPolicy(allowIngress, allowEgress); err != nil {
t.Errorf("TestUpdateNetworkPolicy failed @ UpdateNetworkPolicy")
}
npMgr.Unlock()
}

func TestDeleteNetworkPolicy(t *testing.T) {
Expand Down Expand Up @@ -276,9 +286,11 @@ func TestDeleteNetworkPolicy(t *testing.T) {
},
}

npMgr.Lock()
if err := npMgr.AddNamespace(nsObj); err != nil {
t.Errorf("TestDeleteNetworkPolicy @ npMgr.AddNamespace")
}
npMgr.Unlock()

tcp := corev1.ProtocolTCP
allow := &networkingv1.NetworkPolicy{
Expand All @@ -305,11 +317,13 @@ func TestDeleteNetworkPolicy(t *testing.T) {
},
}

npMgr.Lock()
if err := npMgr.AddNetworkPolicy(allow); err != nil {
t.Errorf("TestAddNetworkPolicy failed @ AddNetworkPolicy")
}

if err := npMgr.DeleteNetworkPolicy(allow); err != nil {
t.Errorf("TestDeleteNetworkPolicy failed @ DeleteNetworkPolicy")
}
npMgr.Unlock()
}
6 changes: 0 additions & 6 deletions npm/pod.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,6 @@ func isSystemPod(podObj *corev1.Pod) bool {

// AddPod handles adding pod ip to its label's ipset.
func (npMgr *NetworkPolicyManager) AddPod(podObj *corev1.Pod) error {
npMgr.Lock()
defer npMgr.Unlock()

if !isValidPod(podObj) {
return nil
}
Expand Down Expand Up @@ -113,9 +110,6 @@ 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 {
npMgr.Lock()
defer npMgr.Unlock()

if !isValidPod(podObj) {
return nil
}
Expand Down
Loading