Revision 6373fdaa9d3e51aff1143dcfbd00a8e13dc43362 authored by Ian Vernon on 23 August 2019, 17:28:30 UTC, committed by Ian Vernon on 23 August 2019, 17:46:49 UTC
Signed-off by: Ian Vernon <ian@cilium.io>
1 parent 9b49d9e
k8s_node.go
// Copyright 2019 Authors of Cilium
//
// 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 main
import (
"context"
"encoding/json"
"strings"
"sync"
"time"
"github.com/cilium/cilium/pkg/controller"
"github.com/cilium/cilium/pkg/k8s"
cilium_v2 "github.com/cilium/cilium/pkg/k8s/apis/cilium.io/v2"
"github.com/cilium/cilium/pkg/k8s/client/clientset/versioned/typed/cilium.io/v2"
"github.com/cilium/cilium/pkg/k8s/utils"
k8sversion "github.com/cilium/cilium/pkg/k8s/version"
"github.com/cilium/cilium/pkg/kvstore/store"
"github.com/cilium/cilium/pkg/node"
nodeStore "github.com/cilium/cilium/pkg/node/store"
"github.com/cilium/cilium/pkg/option"
"github.com/cilium/cilium/pkg/serializer"
"k8s.io/api/core/v1"
core_v1 "k8s.io/api/core/v1"
meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/tools/cache"
)
var (
// kvNodeGCInterval duration for which the nodes are GC in the KVStore.
kvNodeGCInterval time.Duration
)
func runNodeWatcher() error {
log.Info("Starting to synchronize k8s nodes to kvstore...")
serNodes := serializer.NewFunctionQueue(1024)
ciliumNodeStore, err := store.JoinSharedStore(store.Configuration{
Prefix: nodeStore.NodeStorePrefix,
KeyCreator: nodeStore.KeyCreator,
})
if err != nil {
return err
}
k8sNodeStore, nodeController := k8s.NewInformer(
cache.NewListWatchFromClient(k8s.Client().CoreV1().RESTClient(),
"nodes", v1.NamespaceAll, fields.Everything()),
&v1.Node{},
0,
cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
if n := k8s.CopyObjToV1Node(obj); n != nil {
serNodes.Enqueue(func() error {
nodeNew := k8s.ParseNode(n, node.FromKubernetes)
ciliumNodeStore.UpdateKeySync(nodeNew)
return nil
}, serializer.NoRetry)
}
},
UpdateFunc: func(oldObj, newObj interface{}) {
if oldNode := k8s.CopyObjToV1Node(oldObj); oldNode != nil {
if newNode := k8s.CopyObjToV1Node(newObj); newNode != nil {
if k8s.EqualV1Node(oldNode, newNode) {
return
}
serNodes.Enqueue(func() error {
newNode := k8s.ParseNode(newNode, node.FromKubernetes)
ciliumNodeStore.UpdateKeySync(newNode)
return nil
}, serializer.NoRetry)
}
}
},
DeleteFunc: func(obj interface{}) {
n := k8s.CopyObjToV1Node(obj)
if n == nil {
deletedObj, ok := obj.(cache.DeletedFinalStateUnknown)
if !ok {
return
}
// Delete was not observed by the watcher but is
// removed from kube-apiserver. This is the last
// known state and the object no longer exists.
n = k8s.CopyObjToV1Node(deletedObj.Obj)
if n == nil {
return
}
}
serNodes.Enqueue(func() error {
deletedNode := k8s.ParseNode(n, node.FromKubernetes)
ciliumNodeStore.DeleteLocalKey(deletedNode)
return nil
}, serializer.NoRetry)
},
},
k8s.ConvertToNode,
)
go nodeController.Run(wait.NeverStop)
go func() {
cache.WaitForCacheSync(wait.NeverStop, nodeController.HasSynced)
controller.NewManager().UpdateController("kvstore-node-gc",
controller.ControllerParams{
RunInterval: kvNodeGCInterval,
DoFunc: func(_ context.Context) error {
wg := sync.WaitGroup{}
wg.Add(1)
serNodes.Enqueue(func() error {
defer wg.Done()
// Since we serialize all events received from k8s we know that
// at this point the list in k8sNodeStore should be the source of truth
// and we need to delete all nodes in the kvNodeStore that are *not*
// present in the k8sNodeStore.
listOfK8sNodes := k8sNodeStore.ListKeys()
kvStoreNodes := ciliumNodeStore.SharedKeysMap()
for _, k8sNode := range listOfK8sNodes {
// The remaining kvStoreNodes are leftovers
kvStoreNodeName := node.GetKeyNodeName(option.Config.ClusterName, k8sNode)
delete(kvStoreNodes, kvStoreNodeName)
}
for _, kvStoreNode := range kvStoreNodes {
if strings.HasPrefix(kvStoreNode.GetKeyName(), option.Config.ClusterName) {
ciliumNodeStore.DeleteLocalKey(kvStoreNode)
}
}
return nil
}, serializer.NoRetry)
wg.Wait()
return nil
},
})
parallelRequests := 4
removeNodeFromCNP := make(chan func(), 50)
for i := 0; i < parallelRequests; i++ {
go func() {
for f := range removeNodeFromCNP {
f()
}
}()
}
controller.NewManager().UpdateController("cnp-node-gc",
controller.ControllerParams{
RunInterval: kvNodeGCInterval,
DoFunc: func(ctx context.Context) error {
lastRun := time.Now().Add(-kvNodeGCInterval)
k8sCapabilities := k8sversion.Capabilities()
continueID := ""
wg := sync.WaitGroup{}
defer wg.Wait()
kvStoreNodes := ciliumNodeStore.SharedKeysMap()
for {
cnpList, err := ciliumK8sClient.CiliumV2().CiliumNetworkPolicies(core_v1.NamespaceAll).List(meta_v1.ListOptions{
Limit: 10,
Continue: continueID,
})
if err != nil {
return err
}
for _, cnp := range cnpList.Items {
needsUpdate := false
nodesToDelete := map[string]cilium_v2.Timestamp{}
for n, status := range cnp.Status.Nodes {
kvStoreNodeName := node.GetKeyNodeName(option.Config.ClusterName, n)
if _, exists := kvStoreNodes[kvStoreNodeName]; !exists {
// To avoid concurrency issues where a is
// created and adds its CNP Status before the operator
// node watcher receives an event that the node
// was created, we will only delete the node
// from the CNP Status if the last time it was
// update was before the lastRun.
if status.LastUpdated.Before(lastRun) {
nodesToDelete[n] = status.LastUpdated
delete(cnp.Status.Nodes, n)
needsUpdate = true
}
}
}
if needsUpdate {
wg.Add(1)
cnpCpy := cnp.DeepCopy()
removeNodeFromCNP <- func() {
updateCNP(ciliumK8sClient.CiliumV2(), cnpCpy, nodesToDelete, k8sCapabilities)
wg.Done()
}
}
}
continueID = cnpList.Continue
if continueID == "" {
break
}
}
return nil
},
})
}()
return nil
}
func updateCNP(ciliumClient v2.CiliumV2Interface, cnp *cilium_v2.CiliumNetworkPolicy, nodesToDelete map[string]cilium_v2.Timestamp, capabilities k8sversion.ServerCapabilities) {
if len(nodesToDelete) == 0 {
return
}
ns := utils.ExtractNamespace(&cnp.ObjectMeta)
switch {
case capabilities.Patch:
var removeStatusNode, remainingStatusNode []k8s.JSONPatch
for nodeToDelete, timeStamp := range nodesToDelete {
removeStatusNode = append(removeStatusNode,
// It is really unlikely to happen but if a node reappears
// with the same name and updates the CNP Status we will perform
// a test to verify if the lastUpdated timestamp is the same to
// to avoid accidentally deleting that node.
// If any of the nodes fails this test *all* of the JSON patch
// will not be executed.
k8s.JSONPatch{
OP: "test",
Path: "/status/nodes/" + nodeToDelete + "/lastUpdated",
Value: timeStamp,
},
k8s.JSONPatch{
OP: "remove",
Path: "/status/nodes/" + nodeToDelete,
},
)
}
for {
if len(removeStatusNode) > k8s.MaxJSONPatchOperations {
remainingStatusNode = removeStatusNode[k8s.MaxJSONPatchOperations:]
removeStatusNode = removeStatusNode[:k8s.MaxJSONPatchOperations]
}
removeStatusNodeJSON, err := json.Marshal(removeStatusNode)
if err != nil {
break
}
_, err = ciliumClient.CiliumNetworkPolicies(ns).Patch(cnp.GetName(), types.JSONPatchType, removeStatusNodeJSON, "status")
if err != nil {
// We can leave the errors as debug as the GC happens on a best effort
log.WithError(err).Debug("Unable to PATCH")
}
removeStatusNode = remainingStatusNode
if len(remainingStatusNode) == 0 {
return
}
}
case capabilities.UpdateStatus:
// This should be treat is as best effort, we don't care if the
// UpdateStatus fails.
_, err := ciliumClient.CiliumNetworkPolicies(ns).UpdateStatus(cnp)
if err != nil {
// We can leave the errors as debug as the GC happens on a best effort
log.WithError(err).Debug("Unable to UpdateStatus with garbage collected nodes")
}
default:
// This should be treat is as best effort, we don't care if the
// Update fails.
_, err := ciliumClient.CiliumNetworkPolicies(ns).Update(cnp)
if err != nil {
// We can leave the errors as debug as the GC happens on a best effort
log.WithError(err).Debug("Unable to Update CNP with garbage collected nodes")
}
}
}
![swh spinner](/static/img/swh-spinner.gif)
Computing file changes ...