kilo/pkg/k8s/backend.go

230 lines
6.9 KiB
Go
Raw Normal View History

2019-01-18 01:50:10 +00:00
// Copyright 2019 the Kilo authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package k8s
import (
"encoding/json"
"errors"
"fmt"
"net"
"path"
"strings"
"time"
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/strategicpatch"
v1informers "k8s.io/client-go/informers/core/v1"
"k8s.io/client-go/kubernetes"
v1listers "k8s.io/client-go/listers/core/v1"
"k8s.io/client-go/tools/cache"
"github.com/squat/kilo/pkg/mesh"
)
const (
// Backend is the name of this mesh backend.
Backend = "kubernetes"
externalIPAnnotationKey = "kilo.squat.ai/external-ip"
forceExternalIPAnnotationKey = "kilo.squat.ai/force-external-ip"
internalIPAnnotationKey = "kilo.squat.ai/internal-ip"
keyAnnotationKey = "kilo.squat.ai/key"
leaderAnnotationKey = "kilo.squat.ai/leader"
locationAnnotationKey = "kilo.squat.ai/location"
regionLabelKey = "failure-domain.beta.kubernetes.io/region"
jsonPatchSlash = "~1"
jsonRemovePatch = `{"op": "remove", "path": "%s"}`
)
type backend struct {
client kubernetes.Interface
events chan *mesh.Event
informer cache.SharedIndexInformer
lister v1listers.NodeLister
}
// New creates a new instance of a mesh.Backend.
func New(client kubernetes.Interface) mesh.Backend {
informer := v1informers.NewNodeInformer(client, 5*time.Minute, nil)
b := &backend{
client: client,
events: make(chan *mesh.Event),
informer: informer,
lister: v1listers.NewNodeLister(informer.GetIndexer()),
}
return b
}
// CleanUp removes configuration applied to the backend.
func (b *backend) CleanUp(name string) error {
patch := []byte("[" + strings.Join([]string{
fmt.Sprintf(jsonRemovePatch, path.Join("/metadata", "annotations", strings.Replace(externalIPAnnotationKey, "/", jsonPatchSlash, 1))),
fmt.Sprintf(jsonRemovePatch, path.Join("/metadata", "annotations", strings.Replace(internalIPAnnotationKey, "/", jsonPatchSlash, 1))),
fmt.Sprintf(jsonRemovePatch, path.Join("/metadata", "annotations", strings.Replace(keyAnnotationKey, "/", jsonPatchSlash, 1))),
}, ",") + "]")
if _, err := b.client.CoreV1().Nodes().Patch(name, types.JSONPatchType, patch); err != nil {
return fmt.Errorf("failed to patch node: %v", err)
}
return nil
}
// Get gets a single Node by name.
func (b *backend) Get(name string) (*mesh.Node, error) {
n, err := b.lister.Get(name)
if err != nil {
return nil, err
}
return translateNode(n), nil
}
// Init initializes the backend; for this backend that means
// syncing the informer cache.
func (b *backend) Init(stop <-chan struct{}) error {
go b.informer.Run(stop)
if ok := cache.WaitForCacheSync(stop, func() bool {
return b.informer.HasSynced()
}); !ok {
return errors.New("failed to start sync node cache")
}
b.informer.AddEventHandler(
cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
n, ok := obj.(*v1.Node)
if !ok {
// Failed to decode Node; ignoring...
return
}
b.events <- &mesh.Event{Type: mesh.AddEvent, Node: translateNode(n)}
},
UpdateFunc: func(_, obj interface{}) {
n, ok := obj.(*v1.Node)
if !ok {
// Failed to decode Node; ignoring...
return
}
b.events <- &mesh.Event{Type: mesh.UpdateEvent, Node: translateNode(n)}
},
DeleteFunc: func(obj interface{}) {
n, ok := obj.(*v1.Node)
if !ok {
// Failed to decode Node; ignoring...
return
}
b.events <- &mesh.Event{Type: mesh.DeleteEvent, Node: translateNode(n)}
},
},
)
return nil
}
// List gets all the Nodes in the cluster.
func (b *backend) List() ([]*mesh.Node, error) {
ns, err := b.lister.List(labels.Everything())
if err != nil {
return nil, err
}
nodes := make([]*mesh.Node, len(ns))
for i := range ns {
nodes[i] = translateNode(ns[i])
}
return nodes, nil
}
// Set sets the fields of a node.
func (b *backend) Set(name string, node *mesh.Node) error {
old, err := b.lister.Get(name)
if err != nil {
return fmt.Errorf("failed to find node: %v", err)
}
n := old.DeepCopy()
n.ObjectMeta.Annotations[externalIPAnnotationKey] = node.ExternalIP.String()
n.ObjectMeta.Annotations[internalIPAnnotationKey] = node.InternalIP.String()
n.ObjectMeta.Annotations[keyAnnotationKey] = string(node.Key)
oldData, err := json.Marshal(old)
if err != nil {
return err
}
newData, err := json.Marshal(n)
if err != nil {
return err
}
patch, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, v1.Node{})
if err != nil {
return fmt.Errorf("failed to create patch for node %q: %v", n.Name, err)
}
if _, err = b.client.CoreV1().Nodes().Patch(name, types.StrategicMergePatchType, patch); err != nil {
return fmt.Errorf("failed to patch node: %v", err)
}
return nil
}
// Watch returns a chan of node events.
func (b *backend) Watch() <-chan *mesh.Event {
return b.events
}
// translateNode translates a Kubernetes Node to a mesh.Node.
func translateNode(node *v1.Node) *mesh.Node {
if node == nil {
return nil
}
_, subnet, err := net.ParseCIDR(node.Spec.PodCIDR)
// The subnet should only ever fail to parse if the pod CIDR has not been set,
// so in this case set the subnet to nil and let the node be updated.
if err != nil {
subnet = nil
}
_, leader := node.ObjectMeta.Annotations[leaderAnnotationKey]
// Allow the region to be overridden by an explicit location.
location, ok := node.ObjectMeta.Annotations[locationAnnotationKey]
if !ok {
location = node.ObjectMeta.Labels[regionLabelKey]
}
// Allow the external IP to be overridden.
externalIP, ok := node.ObjectMeta.Annotations[forceExternalIPAnnotationKey]
if !ok {
externalIP = node.ObjectMeta.Annotations[externalIPAnnotationKey]
}
return &mesh.Node{
// ExternalIP and InternalIP should only ever fail to parse if the
// remote node's mesh has not yet set its IP address;
// in this case the IP will be nil and
// the mesh can wait for the node to be updated.
ExternalIP: normalizeIP(externalIP),
InternalIP: normalizeIP(node.ObjectMeta.Annotations[internalIPAnnotationKey]),
Key: []byte(node.ObjectMeta.Annotations[keyAnnotationKey]),
Leader: leader,
Location: location,
Name: node.Name,
Subnet: subnet,
}
}
func normalizeIP(ip string) *net.IPNet {
i, ipNet, _ := net.ParseCIDR(ip)
if ipNet == nil {
return ipNet
}
if ip4 := i.To4(); ip4 != nil {
ipNet.IP = ip4
return ipNet
}
ipNet.IP = i.To16()
return ipNet
}