kube.go 9.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304
  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. "k8s.io/kubernetes/pkg/api"
  28. "k8s.io/kubernetes/pkg/client/cache"
  29. clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
  30. "k8s.io/kubernetes/pkg/client/restclient"
  31. "k8s.io/kubernetes/pkg/controller/framework"
  32. "k8s.io/kubernetes/pkg/runtime"
  33. utilruntime "k8s.io/kubernetes/pkg/util/runtime"
  34. "k8s.io/kubernetes/pkg/watch"
  35. )
  36. var (
  37. ErrUnimplemented = errors.New("unimplemented")
  38. kubeSubnetCfg *subnet.Config
  39. )
  40. const (
  41. resyncPeriod = 5 * time.Minute
  42. subnetKubeManagedAnnotation = "flannel.alpha.coreos.com/kube-subnet-manager"
  43. backendDataAnnotation = "flannel.alpha.coreos.com/backend-data"
  44. backendTypeAnnotation = "flannel.alpha.coreos.com/backend-type"
  45. backendPublicIPAnnotation = "flannel.alpha.coreos.com/public-ip"
  46. netConfPath = "/etc/kube-flannel/net-conf.json"
  47. )
  48. type kubeSubnetManager struct {
  49. client clientset.Interface
  50. nodeName string
  51. nodeStore cache.StoreToNodeLister
  52. nodeController *framework.Controller
  53. subnetConf *subnet.Config
  54. events chan subnet.Event
  55. selfEvents chan subnet.Event
  56. }
  57. func NewSubnetManager() (subnet.Manager, error) {
  58. cfg, err := restclient.InClusterConfig()
  59. if err != nil {
  60. return nil, fmt.Errorf("unable to initialize inclusterconfig: %v", err)
  61. }
  62. c, err := clientset.NewForConfig(cfg)
  63. if err != nil {
  64. return nil, fmt.Errorf("unable to initialize client: %v", err)
  65. }
  66. podName := os.Getenv("POD_NAME")
  67. podNamespace := os.Getenv("POD_NAMESPACE")
  68. if podName == "" || podNamespace == "" {
  69. return nil, fmt.Errorf("env variables POD_NAME and POD_NAMESPACE must be set")
  70. }
  71. pod, err := c.Pods(podNamespace).Get(podName)
  72. if err != nil {
  73. return nil, fmt.Errorf("error retrieving pod spec for '%s/%s': %v", podNamespace, podName, err)
  74. }
  75. nodeName := pod.Spec.NodeName
  76. if nodeName == "" {
  77. return nil, fmt.Errorf("node name not present in pod spec '%s/%s'", podNamespace, podName)
  78. }
  79. netConf, err := ioutil.ReadFile(netConfPath)
  80. if err != nil {
  81. return nil, fmt.Errorf("failed to read net conf: %v", err)
  82. }
  83. sc, err := subnet.ParseConfig(string(netConf))
  84. if err != nil {
  85. return nil, fmt.Errorf("error parsing subnet config: %s", err)
  86. }
  87. sm, err := newKubeSubnetManager(c, sc, nodeName)
  88. if err != nil {
  89. return nil, fmt.Errorf("error creating network manager: %s", err)
  90. }
  91. go sm.Run(context.Background())
  92. return sm, err
  93. }
  94. func newKubeSubnetManager(c clientset.Interface, sc *subnet.Config, nodeName string) (*kubeSubnetManager, error) {
  95. var ksm kubeSubnetManager
  96. ksm.client = c
  97. ksm.nodeName = nodeName
  98. ksm.subnetConf = sc
  99. ksm.events = make(chan subnet.Event, 100)
  100. ksm.selfEvents = make(chan subnet.Event, 100)
  101. ksm.nodeStore.Store, ksm.nodeController = framework.NewInformer(
  102. &cache.ListWatch{
  103. ListFunc: func(options api.ListOptions) (runtime.Object, error) {
  104. return ksm.client.Core().Nodes().List(options)
  105. },
  106. WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
  107. return ksm.client.Core().Nodes().Watch(options)
  108. },
  109. },
  110. &api.Node{},
  111. resyncPeriod,
  112. framework.ResourceEventHandlerFuncs{
  113. AddFunc: func(obj interface{}) {
  114. ksm.handleAddLeaseEvent(subnet.EventAdded, obj)
  115. },
  116. UpdateFunc: ksm.handleUpdateLeaseEvent,
  117. DeleteFunc: func(obj interface{}) {
  118. ksm.handleAddLeaseEvent(subnet.EventRemoved, obj)
  119. },
  120. },
  121. )
  122. return &ksm, nil
  123. }
  124. func (ksm *kubeSubnetManager) handleAddLeaseEvent(et subnet.EventType, obj interface{}) {
  125. n := obj.(*api.Node)
  126. if s, ok := n.Annotations[subnetKubeManagedAnnotation]; !ok || s != "true" {
  127. return
  128. }
  129. l, err := nodeToLease(*n)
  130. if err != nil {
  131. glog.Infof("Error turning node %q to lease: %v", n.ObjectMeta.Name, err)
  132. return
  133. }
  134. ksm.events <- subnet.Event{et, l, ""}
  135. if n.ObjectMeta.Name == ksm.nodeName {
  136. ksm.selfEvents <- subnet.Event{et, l, ""}
  137. }
  138. }
  139. func (ksm *kubeSubnetManager) handleUpdateLeaseEvent(oldObj, newObj interface{}) {
  140. o := oldObj.(*api.Node)
  141. n := newObj.(*api.Node)
  142. if s, ok := n.Annotations[subnetKubeManagedAnnotation]; !ok || s != "true" {
  143. return
  144. }
  145. if o.Annotations[backendDataAnnotation] == n.Annotations[backendDataAnnotation] &&
  146. o.Annotations[backendTypeAnnotation] == n.Annotations[backendTypeAnnotation] &&
  147. o.Annotations[backendPublicIPAnnotation] == n.Annotations[backendPublicIPAnnotation] {
  148. return // No change to lease
  149. }
  150. l, err := nodeToLease(*n)
  151. if err != nil {
  152. glog.Infof("Error turning node %q to lease: %v", n.ObjectMeta.Name, err)
  153. return
  154. }
  155. ksm.events <- subnet.Event{subnet.EventAdded, l, ""}
  156. if n.ObjectMeta.Name == ksm.nodeName {
  157. ksm.selfEvents <- subnet.Event{subnet.EventAdded, l, ""}
  158. }
  159. }
  160. func (ksm *kubeSubnetManager) GetNetworkConfig(ctx context.Context, network string) (*subnet.Config, error) {
  161. return ksm.subnetConf, nil
  162. }
  163. func (ksm *kubeSubnetManager) AcquireLease(ctx context.Context, network string, attrs *subnet.LeaseAttrs) (*subnet.Lease, error) {
  164. nobj, found, err := ksm.nodeStore.Store.GetByKey(ksm.nodeName)
  165. if err != nil {
  166. return nil, err
  167. }
  168. if !found {
  169. return nil, fmt.Errorf("node %q not found", ksm.nodeName)
  170. }
  171. n, ok := nobj.(*api.Node)
  172. if !ok {
  173. return nil, fmt.Errorf("nobj was not a *api.Node")
  174. }
  175. if n.Spec.PodCIDR == "" {
  176. return nil, fmt.Errorf("node %q pod cidr not assigned", ksm.nodeName)
  177. }
  178. bd, err := attrs.BackendData.MarshalJSON()
  179. if err != nil {
  180. return nil, err
  181. }
  182. _, cidr, err := net.ParseCIDR(n.Spec.PodCIDR)
  183. if err != nil {
  184. return nil, err
  185. }
  186. if n.Annotations[backendDataAnnotation] != string(bd) ||
  187. n.Annotations[backendTypeAnnotation] != attrs.BackendType ||
  188. n.Annotations[backendPublicIPAnnotation] != attrs.PublicIP.String() ||
  189. n.Annotations[subnetKubeManagedAnnotation] != "true" {
  190. n.Annotations[backendTypeAnnotation] = attrs.BackendType
  191. n.Annotations[backendDataAnnotation] = string(bd)
  192. n.Annotations[backendPublicIPAnnotation] = attrs.PublicIP.String()
  193. n.Annotations[subnetKubeManagedAnnotation] = "true"
  194. n, err = ksm.client.Core().Nodes().Update(n)
  195. if err != nil {
  196. return nil, err
  197. }
  198. }
  199. return &subnet.Lease{
  200. Subnet: ip.FromIPNet(cidr),
  201. Attrs: *attrs,
  202. Expiration: time.Now().Add(24 * time.Hour),
  203. }, nil
  204. }
  205. func (ksm *kubeSubnetManager) RenewLease(ctx context.Context, network string, lease *subnet.Lease) error {
  206. l, err := ksm.AcquireLease(ctx, network, &lease.Attrs)
  207. if err != nil {
  208. return err
  209. }
  210. lease.Subnet = l.Subnet
  211. lease.Attrs = l.Attrs
  212. lease.Expiration = l.Expiration
  213. return nil
  214. }
  215. func (ksm *kubeSubnetManager) WatchLease(ctx context.Context, network string, sn ip.IP4Net, cursor interface{}) (subnet.LeaseWatchResult, error) {
  216. select {
  217. case event := <-ksm.selfEvents:
  218. return subnet.LeaseWatchResult{
  219. Events: []subnet.Event{event},
  220. }, nil
  221. case <-ctx.Done():
  222. return subnet.LeaseWatchResult{}, nil
  223. }
  224. }
  225. func (ksm *kubeSubnetManager) WatchLeases(ctx context.Context, network string, cursor interface{}) (subnet.LeaseWatchResult, error) {
  226. select {
  227. case event := <-ksm.events:
  228. return subnet.LeaseWatchResult{
  229. Events: []subnet.Event{event},
  230. }, nil
  231. case <-ctx.Done():
  232. return subnet.LeaseWatchResult{}, nil
  233. }
  234. }
  235. func (ksm *kubeSubnetManager) WatchNetworks(ctx context.Context, cursor interface{}) (subnet.NetworkWatchResult, error) {
  236. time.Sleep(time.Second)
  237. return subnet.NetworkWatchResult{
  238. Snapshot: []string{""},
  239. }, nil
  240. }
  241. func (ksm *kubeSubnetManager) Run(ctx context.Context) {
  242. defer utilruntime.HandleCrash()
  243. glog.Infof("starting kube subnet manager")
  244. ksm.nodeController.Run(ctx.Done())
  245. }
  246. func nodeToLease(n api.Node) (l subnet.Lease, err error) {
  247. l.Attrs.PublicIP, err = ip.ParseIP4(n.Annotations[backendPublicIPAnnotation])
  248. if err != nil {
  249. return l, err
  250. }
  251. l.Attrs.BackendType = n.Annotations[backendTypeAnnotation]
  252. l.Attrs.BackendData = json.RawMessage(n.Annotations[backendDataAnnotation])
  253. _, cidr, err := net.ParseCIDR(n.Spec.PodCIDR)
  254. if err != nil {
  255. return l, err
  256. }
  257. l.Subnet = ip.FromIPNet(cidr)
  258. l.Expiration = time.Now().Add(24 * time.Hour)
  259. return l, nil
  260. }
  261. // unimplemented
  262. func (ksm *kubeSubnetManager) RevokeLease(ctx context.Context, network string, sn ip.IP4Net) error {
  263. return ErrUnimplemented
  264. }
  265. func (ksm *kubeSubnetManager) AddReservation(ctx context.Context, network string, r *subnet.Reservation) error {
  266. return ErrUnimplemented
  267. }
  268. func (ksm *kubeSubnetManager) RemoveReservation(ctx context.Context, network string, subnet ip.IP4Net) error {
  269. return ErrUnimplemented
  270. }
  271. func (ksm *kubeSubnetManager) ListReservations(ctx context.Context, network string) ([]subnet.Reservation, error) {
  272. return nil, ErrUnimplemented
  273. }