kube.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337
  1. // Copyright 2016 flannel authors
  2. //
  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. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package kube
  15. import (
  16. "encoding/json"
  17. "errors"
  18. "fmt"
  19. "io/ioutil"
  20. "net"
  21. "os"
  22. "time"
  23. "github.com/coreos/flannel/pkg/ip"
  24. "github.com/coreos/flannel/subnet"
  25. "github.com/golang/glog"
  26. "golang.org/x/net/context"
  27. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  28. "k8s.io/apimachinery/pkg/runtime"
  29. "k8s.io/apimachinery/pkg/types"
  30. "k8s.io/apimachinery/pkg/util/strategicpatch"
  31. "k8s.io/apimachinery/pkg/util/wait"
  32. "k8s.io/apimachinery/pkg/watch"
  33. clientset "k8s.io/client-go/kubernetes"
  34. listers "k8s.io/client-go/listers/core/v1"
  35. "k8s.io/client-go/pkg/api"
  36. "k8s.io/client-go/pkg/api/v1"
  37. "k8s.io/client-go/rest"
  38. "k8s.io/client-go/tools/cache"
  39. "k8s.io/client-go/tools/clientcmd"
  40. )
  41. var (
  42. ErrUnimplemented = errors.New("unimplemented")
  43. )
  44. const (
  45. resyncPeriod = 5 * time.Minute
  46. nodeControllerSyncTimeout = 10 * time.Minute
  47. netConfPath = "/etc/kube-flannel/net-conf.json"
  48. )
  49. type kubeSubnetManager struct {
  50. annotations annotations
  51. client clientset.Interface
  52. nodeName string
  53. nodeStore listers.NodeLister
  54. nodeController cache.Controller
  55. subnetConf *subnet.Config
  56. events chan subnet.Event
  57. }
  58. func NewSubnetManager(apiUrl, kubeconfig, prefix string) (subnet.Manager, error) {
  59. var cfg *rest.Config
  60. var err error
  61. // Try to build kubernetes config from a master url or a kubeconfig filepath. If neither masterUrl
  62. // or kubeconfigPath are passed in we fall back to inClusterConfig. If inClusterConfig fails,
  63. // we fallback to the default config.
  64. cfg, err = clientcmd.BuildConfigFromFlags(apiUrl, kubeconfig)
  65. if err != nil {
  66. return nil, fmt.Errorf("fail to create kubernetes config: %v", err)
  67. }
  68. c, err := clientset.NewForConfig(cfg)
  69. if err != nil {
  70. return nil, fmt.Errorf("unable to initialize client: %v", err)
  71. }
  72. // The kube subnet mgr needs to know the k8s node name that it's running on so it can annotate it.
  73. // If we're running as a pod then the POD_NAME and POD_NAMESPACE will be populated and can be used to find the node
  74. // name. Otherwise, the environment variable NODE_NAME can be passed in.
  75. nodeName := os.Getenv("NODE_NAME")
  76. if nodeName == "" {
  77. podName := os.Getenv("POD_NAME")
  78. podNamespace := os.Getenv("POD_NAMESPACE")
  79. if podName == "" || podNamespace == "" {
  80. return nil, fmt.Errorf("env variables POD_NAME and POD_NAMESPACE must be set")
  81. }
  82. pod, err := c.Pods(podNamespace).Get(podName, metav1.GetOptions{})
  83. if err != nil {
  84. return nil, fmt.Errorf("error retrieving pod spec for '%s/%s': %v", podNamespace, podName, err)
  85. }
  86. nodeName = pod.Spec.NodeName
  87. if nodeName == "" {
  88. return nil, fmt.Errorf("node name not present in pod spec '%s/%s'", podNamespace, podName)
  89. }
  90. }
  91. netConf, err := ioutil.ReadFile(netConfPath)
  92. if err != nil {
  93. return nil, fmt.Errorf("failed to read net conf: %v", err)
  94. }
  95. sc, err := subnet.ParseConfig(string(netConf))
  96. if err != nil {
  97. return nil, fmt.Errorf("error parsing subnet config: %s", err)
  98. }
  99. sm, err := newKubeSubnetManager(c, sc, nodeName, prefix)
  100. if err != nil {
  101. return nil, fmt.Errorf("error creating network manager: %s", err)
  102. }
  103. go sm.Run(context.Background())
  104. glog.Infof("Waiting %s for node controller to sync", nodeControllerSyncTimeout)
  105. err = wait.Poll(time.Second, nodeControllerSyncTimeout, func() (bool, error) {
  106. return sm.nodeController.HasSynced(), nil
  107. })
  108. if err != nil {
  109. return nil, fmt.Errorf("error waiting for nodeController to sync state: %v", err)
  110. }
  111. glog.Infof("Node controller sync successful")
  112. return sm, nil
  113. }
  114. func newKubeSubnetManager(c clientset.Interface, sc *subnet.Config, nodeName, prefix string) (*kubeSubnetManager, error) {
  115. var err error
  116. var ksm kubeSubnetManager
  117. ksm.annotations, err = newAnnotations(prefix)
  118. if err != nil {
  119. return nil, err
  120. }
  121. ksm.client = c
  122. ksm.nodeName = nodeName
  123. ksm.subnetConf = sc
  124. ksm.events = make(chan subnet.Event, 5000)
  125. indexer, controller := cache.NewIndexerInformer(
  126. &cache.ListWatch{
  127. ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
  128. return ksm.client.CoreV1().Nodes().List(options)
  129. },
  130. WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
  131. return ksm.client.CoreV1().Nodes().Watch(options)
  132. },
  133. },
  134. &v1.Node{},
  135. resyncPeriod,
  136. cache.ResourceEventHandlerFuncs{
  137. AddFunc: func(obj interface{}) {
  138. ksm.handleAddLeaseEvent(subnet.EventAdded, obj)
  139. },
  140. UpdateFunc: ksm.handleUpdateLeaseEvent,
  141. DeleteFunc: func(obj interface{}) {
  142. node, isNode := obj.(*v1.Node)
  143. // We can get DeletedFinalStateUnknown instead of *api.Node here and we need to handle that correctly.
  144. if !isNode {
  145. deletedState, ok := obj.(cache.DeletedFinalStateUnknown)
  146. if !ok {
  147. glog.Infof("Error received unexpected object: %v", obj)
  148. return
  149. }
  150. node, ok = deletedState.Obj.(*v1.Node)
  151. if !ok {
  152. glog.Infof("Error deletedFinalStateUnknown contained non-Node object: %v", deletedState.Obj)
  153. return
  154. }
  155. obj = node
  156. }
  157. ksm.handleAddLeaseEvent(subnet.EventRemoved, obj)
  158. },
  159. },
  160. cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
  161. )
  162. ksm.nodeController = controller
  163. ksm.nodeStore = listers.NewNodeLister(indexer)
  164. return &ksm, nil
  165. }
  166. func (ksm *kubeSubnetManager) handleAddLeaseEvent(et subnet.EventType, obj interface{}) {
  167. n := obj.(*v1.Node)
  168. if s, ok := n.Annotations[ksm.annotations.SubnetKubeManaged]; !ok || s != "true" {
  169. return
  170. }
  171. l, err := ksm.nodeToLease(*n)
  172. if err != nil {
  173. glog.Infof("Error turning node %q to lease: %v", n.ObjectMeta.Name, err)
  174. return
  175. }
  176. ksm.events <- subnet.Event{et, l}
  177. }
  178. func (ksm *kubeSubnetManager) handleUpdateLeaseEvent(oldObj, newObj interface{}) {
  179. o := oldObj.(*v1.Node)
  180. n := newObj.(*v1.Node)
  181. if s, ok := n.Annotations[ksm.annotations.SubnetKubeManaged]; !ok || s != "true" {
  182. return
  183. }
  184. if o.Annotations[ksm.annotations.BackendData] == n.Annotations[ksm.annotations.BackendData] &&
  185. o.Annotations[ksm.annotations.BackendType] == n.Annotations[ksm.annotations.BackendType] &&
  186. o.Annotations[ksm.annotations.BackendPublicIP] == n.Annotations[ksm.annotations.BackendPublicIP] {
  187. return // No change to lease
  188. }
  189. l, err := ksm.nodeToLease(*n)
  190. if err != nil {
  191. glog.Infof("Error turning node %q to lease: %v", n.ObjectMeta.Name, err)
  192. return
  193. }
  194. ksm.events <- subnet.Event{subnet.EventAdded, l}
  195. }
  196. func (ksm *kubeSubnetManager) GetNetworkConfig(ctx context.Context) (*subnet.Config, error) {
  197. return ksm.subnetConf, nil
  198. }
  199. func (ksm *kubeSubnetManager) AcquireLease(ctx context.Context, attrs *subnet.LeaseAttrs) (*subnet.Lease, error) {
  200. cachedNode, err := ksm.nodeStore.Get(ksm.nodeName)
  201. if err != nil {
  202. return nil, err
  203. }
  204. nobj, err := api.Scheme.DeepCopy(cachedNode)
  205. if err != nil {
  206. return nil, err
  207. }
  208. n := nobj.(*v1.Node)
  209. if n.Spec.PodCIDR == "" {
  210. return nil, fmt.Errorf("node %q pod cidr not assigned", ksm.nodeName)
  211. }
  212. bd, err := attrs.BackendData.MarshalJSON()
  213. if err != nil {
  214. return nil, err
  215. }
  216. _, cidr, err := net.ParseCIDR(n.Spec.PodCIDR)
  217. if err != nil {
  218. return nil, err
  219. }
  220. if n.Annotations[ksm.annotations.BackendData] != string(bd) ||
  221. n.Annotations[ksm.annotations.BackendType] != attrs.BackendType ||
  222. n.Annotations[ksm.annotations.BackendPublicIP] != attrs.PublicIP.String() ||
  223. n.Annotations[ksm.annotations.SubnetKubeManaged] != "true" ||
  224. (n.Annotations[ksm.annotations.BackendPublicIPOverwrite] != "" && n.Annotations[ksm.annotations.BackendPublicIPOverwrite] != attrs.PublicIP.String()) {
  225. n.Annotations[ksm.annotations.BackendType] = attrs.BackendType
  226. n.Annotations[ksm.annotations.BackendData] = string(bd)
  227. if n.Annotations[ksm.annotations.BackendPublicIPOverwrite] != "" {
  228. if n.Annotations[ksm.annotations.BackendPublicIP] != n.Annotations[ksm.annotations.BackendPublicIPOverwrite] {
  229. glog.Infof("Overriding public ip with '%s' from node annotation '%s'",
  230. n.Annotations[ksm.annotations.BackendPublicIPOverwrite],
  231. ksm.annotations.BackendPublicIPOverwrite)
  232. n.Annotations[ksm.annotations.BackendPublicIP] = n.Annotations[ksm.annotations.BackendPublicIPOverwrite]
  233. }
  234. } else {
  235. n.Annotations[ksm.annotations.BackendPublicIP] = attrs.PublicIP.String()
  236. }
  237. n.Annotations[ksm.annotations.SubnetKubeManaged] = "true"
  238. oldData, err := json.Marshal(cachedNode)
  239. if err != nil {
  240. return nil, err
  241. }
  242. newData, err := json.Marshal(n)
  243. if err != nil {
  244. return nil, err
  245. }
  246. patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, v1.Node{})
  247. if err != nil {
  248. return nil, fmt.Errorf("failed to create patch for node %q: %v", ksm.nodeName, err)
  249. }
  250. _, err = ksm.client.CoreV1().Nodes().Patch(ksm.nodeName, types.StrategicMergePatchType, patchBytes, "status")
  251. if err != nil {
  252. return nil, err
  253. }
  254. }
  255. return &subnet.Lease{
  256. Subnet: ip.FromIPNet(cidr),
  257. Attrs: *attrs,
  258. Expiration: time.Now().Add(24 * time.Hour),
  259. }, nil
  260. }
  261. func (ksm *kubeSubnetManager) WatchLeases(ctx context.Context, cursor interface{}) (subnet.LeaseWatchResult, error) {
  262. select {
  263. case event := <-ksm.events:
  264. return subnet.LeaseWatchResult{
  265. Events: []subnet.Event{event},
  266. }, nil
  267. case <-ctx.Done():
  268. return subnet.LeaseWatchResult{}, nil
  269. }
  270. }
  271. func (ksm *kubeSubnetManager) Run(ctx context.Context) {
  272. glog.Infof("Starting kube subnet manager")
  273. ksm.nodeController.Run(ctx.Done())
  274. }
  275. func (ksm *kubeSubnetManager) nodeToLease(n v1.Node) (l subnet.Lease, err error) {
  276. l.Attrs.PublicIP, err = ip.ParseIP4(n.Annotations[ksm.annotations.BackendPublicIP])
  277. if err != nil {
  278. return l, err
  279. }
  280. l.Attrs.BackendType = n.Annotations[ksm.annotations.BackendType]
  281. l.Attrs.BackendData = json.RawMessage(n.Annotations[ksm.annotations.BackendData])
  282. _, cidr, err := net.ParseCIDR(n.Spec.PodCIDR)
  283. if err != nil {
  284. return l, err
  285. }
  286. l.Subnet = ip.FromIPNet(cidr)
  287. return l, nil
  288. }
  289. // RenewLease: unimplemented
  290. func (ksm *kubeSubnetManager) RenewLease(ctx context.Context, lease *subnet.Lease) error {
  291. return ErrUnimplemented
  292. }
  293. func (ksm *kubeSubnetManager) WatchLease(ctx context.Context, sn ip.IP4Net, cursor interface{}) (subnet.LeaseWatchResult, error) {
  294. return subnet.LeaseWatchResult{}, ErrUnimplemented
  295. }
  296. func (ksm *kubeSubnetManager) Name() string {
  297. return fmt.Sprintf("Kubernetes Subnet Manager - %s", ksm.nodeName)
  298. }