Add mutex when searching replicaset's owner (#230)

* fix: add mux durning searching replicaSet's Owner Signed-off-by: niejiangang <niejiangang@harmonycloud.cn> * test: add a testcase named OnAddPodWhileReplicaSetUpdating link #229 Signed-off-by: niejiangang <niejiangang@harmonycloud.cn> * test: a simple mistake in test case Signed-off-by: niejiangang <niejiangang@harmonycloud.cn>
KindlingProject · May 31, 2022 · 8d2fec6 · 8d2fec6
1 parent 63bd892
commit 8d2fec6
Show file tree

Hide file tree

Showing 5 changed files with 55 additions and 5 deletions.
diff --git a/collector/go.mod b/collector/go.mod
@@ -15,7 +15,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.0 // indirect
 	github.com/spf13/viper v1.10.1
-	github.com/stretchr/testify v1.7.0
+	github.com/stretchr/testify v1.7.1
 	github.com/vishvananda/netns v0.0.0-20211101163701-50045581ed74
 	go.opentelemetry.io/otel v1.2.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.25.0

diff --git a/collector/go.sum b/collector/go.sum
@@ -467,6 +467,8 @@ github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/subosito/gotenv v1.2.0 h1:Slr1R9HxAlEKefgq5jn9U+DnETlIUa6HfgEzj0g5d7s=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=

diff --git a/collector/metadata/kubernetes/pod_watch.go b/collector/metadata/kubernetes/pod_watch.go
@@ -138,6 +138,7 @@ func onAdd(obj interface{}) {
 	workloadTypeTmp := ""
 	workloadNameTmp := ""
 
+	rsUpdateMutex.RLock()
 	for _, owner := range pod.OwnerReferences {
 		// only care about the controller
 		if owner.Controller == nil || *owner.Controller != true {
@@ -159,6 +160,7 @@ func onAdd(obj interface{}) {
 		workloadNameTmp = owner.Name
 		break
 	}
+	rsUpdateMutex.RUnlock()
 
 	serviceInfoSlice := globalServiceInfo.GetServiceMatchLabels(pI.Namespace, pI.Labels)
 	var serviceInfo *K8sServiceInfo

diff --git a/collector/metadata/kubernetes/pod_watch_test.go b/collector/metadata/kubernetes/pod_watch_test.go
@@ -1,9 +1,11 @@
 package kubernetes
 
 import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"testing"
 )
 
 func TestTruncateContainerId(t *testing.T) {
@@ -49,6 +51,49 @@ func TestOnAdd(t *testing.T) {
 	t.Log(MetaDataCache)
 }
 
+// ISSUE https://github.com/CloudDectective-Harmonycloud/kindling/issues/229
+func TestOnAddPodWhileReplicaSetUpdating(t *testing.T) {
+	globalPodInfo = &podMap{
+		Info: make(map[string]map[string]*PodInfo),
+	}
+	globalServiceInfo = &ServiceMap{
+		ServiceMap: make(map[string]map[string]*K8sServiceInfo),
+	}
+	globalRsInfo = &ReplicaSetMap{
+		Info: make(map[string]Controller),
+	}
+	// Firstly deployment created and add old RS and old POD
+	controller := true
+	oldRs := CreateReplicaSet()
+	oldRs.SetResourceVersion("old")
+	newRs := CreateReplicaSet()
+	newRs.SetResourceVersion("new")
+	oldPOD := CreatePod(true)
+	oldPOD.SetResourceVersion("old")
+	oldPOD.OwnerReferences[0].Controller = &controller
+	newPOD := CreatePod(true)
+	newPOD.SetResourceVersion("new")
+	newPOD.OwnerReferences[0].Controller = &controller
+	onAddReplicaSet(oldRs)
+	onAdd(oldPOD)
+
+	// Secondly POD&RS were been updated
+
+	go func() {
+		for i := 0; i < 1000; i++ {
+			OnUpdateReplicaSet(oldRs, newRs)
+		}
+	}()
+
+	for i := 0; i < 100; i++ {
+		OnUpdate(oldPOD, newPOD)
+		// Thirdly check the pod's workload_kind
+		pod, ok := MetaDataCache.GetPodByContainerId(TruncateContainerId(newPOD.Status.ContainerStatuses[0].ContainerID))
+		require.True(t, ok, "failed to get target POD")
+		require.Equal(t, "deployment", pod.WorkloadKind, "failed to get the real workload_kind")
+	}
+}
+
 func TestOnAddLowercaseWorkload(t *testing.T) {
 	globalPodInfo = &podMap{
 		Info: make(map[string]map[string]*PodInfo),

diff --git a/collector/metadata/kubernetes/replicaset_watch.go b/collector/metadata/kubernetes/replicaset_watch.go
@@ -2,6 +2,9 @@ package kubernetes
 
 import (
 	"fmt"
+	_ "path/filepath"
+	"sync"
+
 	appv1 "k8s.io/api/apps/v1"
 	_ "k8s.io/apimachinery/pkg/apis/meta/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -11,8 +14,6 @@ import (
 	"k8s.io/client-go/tools/cache"
 	_ "k8s.io/client-go/tools/clientcmd"
 	_ "k8s.io/client-go/util/homedir"
-	_ "path/filepath"
-	"sync"
 )
 
 const ReplicaSetKind = "ReplicaSet"
@@ -24,7 +25,7 @@ type ReplicaSetMap struct {
 }
 
 var globalRsInfo = newReplicaSetMap()
-var rsUpdateMutex sync.Mutex
+var rsUpdateMutex sync.RWMutex
 
 type Controller struct {
 	Name       string