github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Fix potential data race

pull/6/head
Random-Liu 2015-12-10 10:35:02 -08:00
parent 0bf11d89cf
commit eb551f388c
1 changed files with 27 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
pkg/kubelet/config/config.go
View File

@ -194,9 +194,9 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
s.podLock.Lock() s.podLock.Lock()
defer s.podLock.Unlock() defer s.podLock.Unlock()
adds = &kubetypes.PodUpdate{Op: kubetypes.ADD, Source: source} addPods := []*api.Pod{}
updates = &kubetypes.PodUpdate{Op: kubetypes.UPDATE, Source: source} updatePods := []*api.Pod{}
deletes = &kubetypes.PodUpdate{Op: kubetypes.REMOVE, Source: source} deletePods := []*api.Pod{}
pods := s.pods[source] pods := s.pods[source]
if pods == nil { if pods == nil {
@ -223,7 +223,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
if existing, found := pods[name]; found { if existing, found := pods[name]; found {
if checkAndUpdatePod(existing, ref) { if checkAndUpdatePod(existing, ref) {
// this is an update // this is an update
updates.Pods = append(updates.Pods, existing) updatePods = append(updatePods, existing)
continue continue
} }
// this is a no-op // this is a no-op
@ -232,7 +232,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
// this is an add // this is an add
recordFirstSeenTime(ref) recordFirstSeenTime(ref)
pods[name] = ref pods[name] = ref
adds.Pods = append(adds.Pods, ref) addPods = append(addPods, ref)
} }
case kubetypes.REMOVE: case kubetypes.REMOVE:
@ -242,7 +242,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
if existing, found := pods[name]; found { if existing, found := pods[name]; found {
// this is a delete // this is a delete
delete(pods, name) delete(pods, name)
deletes.Pods = append(deletes.Pods, existing) deletePods = append(deletePods, existing)
continue continue
} }
// this is a no-op // this is a no-op
@ -267,7 +267,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
pods[name] = existing pods[name] = existing
if checkAndUpdatePod(existing, ref) { if checkAndUpdatePod(existing, ref) {
// this is an update // this is an update
updates.Pods = append(updates.Pods, existing) updatePods = append(updatePods, existing)
continue continue
} }
// this is a no-op // this is a no-op
@ -275,13 +275,13 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
} }
recordFirstSeenTime(ref) recordFirstSeenTime(ref)
pods[name] = ref pods[name] = ref
adds.Pods = append(adds.Pods, ref) addPods = append(addPods, ref)
} }
for name, existing := range oldPods { for name, existing := range oldPods {
if _, found := pods[name]; !found { if _, found := pods[name]; !found {
// this is a delete // this is a delete
deletes.Pods = append(deletes.Pods, existing) deletePods = append(deletePods, existing)
} }
} }
@ -291,6 +291,11 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
} }
s.pods[source] = pods s.pods[source] = pods
adds = &kubetypes.PodUpdate{Op: kubetypes.ADD, Pods: copyPods(addPods), Source: source}
updates = &kubetypes.PodUpdate{Op: kubetypes.UPDATE, Pods: copyPods(updatePods), Source: source}
deletes = &kubetypes.PodUpdate{Op: kubetypes.REMOVE, Pods: copyPods(deletePods), Source: source}
return adds, updates, deletes return adds, updates, deletes
} }
@ -467,3 +472,16 @@ func bestPodIdentString(pod *api.Pod) string {
} }
return fmt.Sprintf("%s.%s", name, namespace) return fmt.Sprintf("%s.%s", name, namespace)
} }
func copyPods(sourcePods []*api.Pod) []*api.Pod {
pods := []*api.Pod{}
for _, source := range sourcePods {
// Use a deep copy here just in case
pod, err := api.Scheme.Copy(source)
if err != nil {
glog.Errorf("unable to copy pod: %v", err)
}
pods = append(pods, pod.(*api.Pod))
}
return pods
}