pod_manager.go 9.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298
  1. /*
  2. Copyright 2015 The Kubernetes Authors.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. package pod
  14. import (
  15. "sync"
  16. "k8s.io/kubernetes/pkg/api"
  17. kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
  18. "k8s.io/kubernetes/pkg/types"
  19. )
  20. // Manager stores and manages access to pods, maintaining the mappings
  21. // between static pods and mirror pods.
  22. //
  23. // The kubelet discovers pod updates from 3 sources: file, http, and
  24. // apiserver. Pods from non-apiserver sources are called static pods, and API
  25. // server is not aware of the existence of static pods. In order to monitor
  26. // the status of such pods, the kubelet creates a mirror pod for each static
  27. // pod via the API server.
  28. //
  29. // A mirror pod has the same pod full name (name and namespace) as its static
  30. // counterpart (albeit different metadata such as UID, etc). By leveraging the
  31. // fact that the kubelet reports the pod status using the pod full name, the
  32. // status of the mirror pod always reflects the actual status of the static
  33. // pod. When a static pod gets deleted, the associated orphaned mirror pod
  34. // will also be removed.
  35. type Manager interface {
  36. // GetPods returns the regular pods bound to the kubelet and their spec.
  37. GetPods() []*api.Pod
  38. // GetPodByName returns the (non-mirror) pod that matches full name, as well as
  39. // whether the pod was found.
  40. GetPodByFullName(podFullName string) (*api.Pod, bool)
  41. // GetPodByName provides the (non-mirror) pod that matches namespace and
  42. // name, as well as whether the pod was found.
  43. GetPodByName(namespace, name string) (*api.Pod, bool)
  44. // GetPodByUID provides the (non-mirror) pod that matches pod UID, as well as
  45. // whether the pod is found.
  46. GetPodByUID(types.UID) (*api.Pod, bool)
  47. // GetPodByMirrorPod returns the static pod for the given mirror pod and
  48. // whether it was known to the pod manger.
  49. GetPodByMirrorPod(*api.Pod) (*api.Pod, bool)
  50. // GetMirrorPodByPod returns the mirror pod for the given static pod and
  51. // whether it was known to the pod manager.
  52. GetMirrorPodByPod(*api.Pod) (*api.Pod, bool)
  53. // GetPodsAndMirrorPods returns the both regular and mirror pods.
  54. GetPodsAndMirrorPods() ([]*api.Pod, []*api.Pod)
  55. // SetPods replaces the internal pods with the new pods.
  56. // It is currently only used for testing.
  57. SetPods(pods []*api.Pod)
  58. // AddPod adds the given pod to the manager.
  59. AddPod(pod *api.Pod)
  60. // UpdatePod updates the given pod in the manager.
  61. UpdatePod(pod *api.Pod)
  62. // DeletePod deletes the given pod from the manager. For mirror pods,
  63. // this means deleting the mappings related to mirror pods. For non-
  64. // mirror pods, this means deleting from indexes for all non-mirror pods.
  65. DeletePod(pod *api.Pod)
  66. // DeleteOrphanedMirrorPods deletes all mirror pods which do not have
  67. // associated static pods. This method sends deletion requests to the API
  68. // server, but does NOT modify the internal pod storage in basicManager.
  69. DeleteOrphanedMirrorPods()
  70. // TranslatePodUID returns the UID which is the mirror pod or static pod
  71. // of the pod with the given UID. If the UID belongs to a mirror pod,
  72. // returns the UID of its static pod. If the UID belongs to a static pod,
  73. // returns the UID of its mirror pod. Otherwise, returns the original UID.
  74. //
  75. // All public-facing functions should perform this translation for UIDs
  76. // because user may provide a mirror pod UID, which is not recognized by
  77. // internal Kubelet functions.
  78. TranslatePodUID(uid types.UID) types.UID
  79. // GetUIDTranslations returns the mappings of static pod UIDs to mirror pod
  80. // UIDs and mirror pod UIDs to static pod UIDs.
  81. GetUIDTranslations() (podToMirror, mirrorToPod map[types.UID]types.UID)
  82. // IsMirrorPodOf returns true if mirrorPod is a correct representation of
  83. // pod; false otherwise.
  84. IsMirrorPodOf(mirrorPod, pod *api.Pod) bool
  85. MirrorClient
  86. }
  87. // basicManager is a functional Manger.
  88. //
  89. // All fields in basicManager are read-only and are updated calling SetPods,
  90. // AddPod, UpdatePod, or DeletePod.
  91. type basicManager struct {
  92. // Protects all internal maps.
  93. lock sync.RWMutex
  94. // Regular pods indexed by UID.
  95. podByUID map[types.UID]*api.Pod
  96. // Mirror pods indexed by UID.
  97. mirrorPodByUID map[types.UID]*api.Pod
  98. // Pods indexed by full name for easy access.
  99. podByFullName map[string]*api.Pod
  100. mirrorPodByFullName map[string]*api.Pod
  101. // Mirror pod UID to pod UID map.
  102. translationByUID map[types.UID]types.UID
  103. // A mirror pod client to create/delete mirror pods.
  104. MirrorClient
  105. }
  106. // NewBasicPodManager returns a functional Manager.
  107. func NewBasicPodManager(client MirrorClient) Manager {
  108. pm := &basicManager{}
  109. pm.MirrorClient = client
  110. pm.SetPods(nil)
  111. return pm
  112. }
  113. // Set the internal pods based on the new pods.
  114. func (pm *basicManager) SetPods(newPods []*api.Pod) {
  115. pm.lock.Lock()
  116. defer pm.lock.Unlock()
  117. pm.podByUID = make(map[types.UID]*api.Pod)
  118. pm.podByFullName = make(map[string]*api.Pod)
  119. pm.mirrorPodByUID = make(map[types.UID]*api.Pod)
  120. pm.mirrorPodByFullName = make(map[string]*api.Pod)
  121. pm.translationByUID = make(map[types.UID]types.UID)
  122. pm.updatePodsInternal(newPods...)
  123. }
  124. func (pm *basicManager) AddPod(pod *api.Pod) {
  125. pm.UpdatePod(pod)
  126. }
  127. func (pm *basicManager) UpdatePod(pod *api.Pod) {
  128. pm.lock.Lock()
  129. defer pm.lock.Unlock()
  130. pm.updatePodsInternal(pod)
  131. }
  132. // updatePodsInternal replaces the given pods in the current state of the
  133. // manager, updating the various indices. The caller is assumed to hold the
  134. // lock.
  135. func (pm *basicManager) updatePodsInternal(pods ...*api.Pod) {
  136. for _, pod := range pods {
  137. podFullName := kubecontainer.GetPodFullName(pod)
  138. if IsMirrorPod(pod) {
  139. pm.mirrorPodByUID[pod.UID] = pod
  140. pm.mirrorPodByFullName[podFullName] = pod
  141. if p, ok := pm.podByFullName[podFullName]; ok {
  142. pm.translationByUID[pod.UID] = p.UID
  143. }
  144. } else {
  145. pm.podByUID[pod.UID] = pod
  146. pm.podByFullName[podFullName] = pod
  147. if mirror, ok := pm.mirrorPodByFullName[podFullName]; ok {
  148. pm.translationByUID[mirror.UID] = pod.UID
  149. }
  150. }
  151. }
  152. }
  153. func (pm *basicManager) DeletePod(pod *api.Pod) {
  154. pm.lock.Lock()
  155. defer pm.lock.Unlock()
  156. podFullName := kubecontainer.GetPodFullName(pod)
  157. if IsMirrorPod(pod) {
  158. delete(pm.mirrorPodByUID, pod.UID)
  159. delete(pm.mirrorPodByFullName, podFullName)
  160. delete(pm.translationByUID, pod.UID)
  161. } else {
  162. delete(pm.podByUID, pod.UID)
  163. delete(pm.podByFullName, podFullName)
  164. }
  165. }
  166. func (pm *basicManager) GetPods() []*api.Pod {
  167. pm.lock.RLock()
  168. defer pm.lock.RUnlock()
  169. return podsMapToPods(pm.podByUID)
  170. }
  171. func (pm *basicManager) GetPodsAndMirrorPods() ([]*api.Pod, []*api.Pod) {
  172. pm.lock.RLock()
  173. defer pm.lock.RUnlock()
  174. pods := podsMapToPods(pm.podByUID)
  175. mirrorPods := podsMapToPods(pm.mirrorPodByUID)
  176. return pods, mirrorPods
  177. }
  178. func (pm *basicManager) GetPodByUID(uid types.UID) (*api.Pod, bool) {
  179. pm.lock.RLock()
  180. defer pm.lock.RUnlock()
  181. pod, ok := pm.podByUID[uid]
  182. return pod, ok
  183. }
  184. func (pm *basicManager) GetPodByName(namespace, name string) (*api.Pod, bool) {
  185. podFullName := kubecontainer.BuildPodFullName(name, namespace)
  186. return pm.GetPodByFullName(podFullName)
  187. }
  188. func (pm *basicManager) GetPodByFullName(podFullName string) (*api.Pod, bool) {
  189. pm.lock.RLock()
  190. defer pm.lock.RUnlock()
  191. pod, ok := pm.podByFullName[podFullName]
  192. return pod, ok
  193. }
  194. func (pm *basicManager) TranslatePodUID(uid types.UID) types.UID {
  195. if uid == "" {
  196. return uid
  197. }
  198. pm.lock.RLock()
  199. defer pm.lock.RUnlock()
  200. if translated, ok := pm.translationByUID[uid]; ok {
  201. return translated
  202. }
  203. return uid
  204. }
  205. func (pm *basicManager) GetUIDTranslations() (podToMirror, mirrorToPod map[types.UID]types.UID) {
  206. pm.lock.RLock()
  207. defer pm.lock.RUnlock()
  208. podToMirror = make(map[types.UID]types.UID, len(pm.translationByUID))
  209. mirrorToPod = make(map[types.UID]types.UID, len(pm.translationByUID))
  210. for k, v := range pm.translationByUID {
  211. mirrorToPod[k] = v
  212. podToMirror[v] = k
  213. }
  214. return podToMirror, mirrorToPod
  215. }
  216. func (pm *basicManager) getOrphanedMirrorPodNames() []string {
  217. pm.lock.RLock()
  218. defer pm.lock.RUnlock()
  219. var podFullNames []string
  220. for podFullName := range pm.mirrorPodByFullName {
  221. if _, ok := pm.podByFullName[podFullName]; !ok {
  222. podFullNames = append(podFullNames, podFullName)
  223. }
  224. }
  225. return podFullNames
  226. }
  227. func (pm *basicManager) DeleteOrphanedMirrorPods() {
  228. podFullNames := pm.getOrphanedMirrorPodNames()
  229. for _, podFullName := range podFullNames {
  230. pm.MirrorClient.DeleteMirrorPod(podFullName)
  231. }
  232. }
  233. func (pm *basicManager) IsMirrorPodOf(mirrorPod, pod *api.Pod) bool {
  234. // Check name and namespace first.
  235. if pod.Name != mirrorPod.Name || pod.Namespace != mirrorPod.Namespace {
  236. return false
  237. }
  238. hash, ok := getHashFromMirrorPod(mirrorPod)
  239. if !ok {
  240. return false
  241. }
  242. return hash == getPodHash(pod)
  243. }
  244. func podsMapToPods(UIDMap map[types.UID]*api.Pod) []*api.Pod {
  245. pods := make([]*api.Pod, 0, len(UIDMap))
  246. for _, pod := range UIDMap {
  247. pods = append(pods, pod)
  248. }
  249. return pods
  250. }
  251. func (pm *basicManager) GetMirrorPodByPod(pod *api.Pod) (*api.Pod, bool) {
  252. pm.lock.RLock()
  253. defer pm.lock.RUnlock()
  254. mirrorPod, ok := pm.mirrorPodByFullName[kubecontainer.GetPodFullName(pod)]
  255. return mirrorPod, ok
  256. }
  257. func (pm *basicManager) GetPodByMirrorPod(mirrorPod *api.Pod) (*api.Pod, bool) {
  258. pm.lock.RLock()
  259. defer pm.lock.RUnlock()
  260. pod, ok := pm.podByFullName[kubecontainer.GetPodFullName(mirrorPod)]
  261. return pod, ok
  262. }