func (c *customK8sNodeResourceClient) extractResourcesFromAnnotation(node *apiv1.Node, name string) ([]*model.KVPair, error) {
	logContext := log.WithFields(log.Fields{
		"ResourceType": name,
		"Resource":     c.ResourceType,
		"Namespace":    c.Namespace,
	})
	logContext.Debug("Extract node and resource info from Key")
	resStrings := make(map[string]string, 0)
	resNames := []string{}
	if name != "" {
		ak := c.nameToAnnotationKey(name)
		if v, ok := node.Annotations[ak]; ok {
			resStrings[name] = v
			resNames = append(resNames, name)
		}
	} else {
		for ak, v := range node.Annotations {
			if n := c.annotationKeyToName(ak); n != "" {
				resStrings[n] = v
				resNames = append(resNames, n)
			}
		}
	}
	sort.Strings(resNames)
	kvps := []*model.KVPair{}
	for _, resName := range resNames {
		key, err := c.Converter.NodeAndNameToKey(node.GetName(), resName)
		if err != nil {
			logContext.WithField("ResourceType", resName).WithError(err).Error("Error unmarshalling key")
			if name != "" {
				return nil, err
			}
			continue
		}
		value, err := model.ParseValue(key, []byte(resStrings[resName]))
		if err != nil {
			logContext.WithField("ResourceType", resName).WithError(err).Error("Error unmarshalling value")
			if name != "" {
				return nil, err
			}
			continue
		}
		kvp := &model.KVPair{
			Key:      key,
			Value:    value,
			Revision: node.GetResourceVersion(),
		}
		kvps = append(kvps, kvp)
	}
	return kvps, nil
}
extractResourcesFromAnnotation queries the current Kubernetes Node resource
and parses the per-node resource entries configured in the annotations.
Returns the Node resource configuration and the slice of parsed resources.
package resources
import (
	"context"
	"sort"
	"strings"
	"github.com/projectcalico/libcalico-go/lib/backend/model"
	"github.com/projectcalico/libcalico-go/lib/errors"
	log "github.com/sirupsen/logrus"
	apiv1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
)
type action string
const (
	actApply  action = "Apply"
	actCreate action = "Create"
	actUpdate action = "Update"
)
type CustomK8sNodeResourceConverter interface {
	ListInterfaceToNodeAndName(model.ListInterface) (string, string, error)
	KeyToNodeAndName(model.Key) (string, string, error)
	NodeAndNameToKey(string, string) (model.Key, error)
}
type CustomK8sNodeResourceClientConfig struct {
	ClientSet    *kubernetes.Clientset
	ResourceType string
	Converter    CustomK8sNodeResourceConverter
	Namespace    string
}
func NewCustomK8sNodeResourceClient(config CustomK8sNodeResourceClientConfig) K8sResourceClient {
	return &nodeRetryWrapper{
		retryWrapper: &retryWrapper{
			client: &customK8sNodeResourceClient{
				CustomK8sNodeResourceClientConfig: config,
				annotationKeyPrefix:               config.Namespace + "/",
			},
		},
	}
}
type nodeRetryWrapper struct {
	*retryWrapper
}
type customK8sNodeResourceClient struct {
	CustomK8sNodeResourceClientConfig
	annotationKeyPrefix string
}
func (c *customK8sNodeResourceClient) Get(key model.Key) (*model.KVPair, error) {
	logContext := log.WithFields(log.Fields{
		"Key":       key,
		"Resource":  c.ResourceType,
		"Namespace": c.Namespace,
	})
	logContext.Debug("Get per-Node resource")
	name, node, err := c.getNameAndNodeFromKey(key)
	if err != nil {
		logContext.WithError(err).Info("Failed to get resource")
		return nil, err
	}
	kvps, err := c.extractResourcesFromAnnotation(node, name)
	if err != nil {
		logContext.WithError(err).Error("Failed to get resource: error in data")
		return nil, err
	}
	if len(kvps) != 1 {
		logContext.Warning("Failed to get resource: resource does not exist")
		return nil, errors.ErrorResourceDoesNotExist{Identifier: key}
	}
	return kvps[0], nil
}
func (c *customK8sNodeResourceClient) List(list model.ListInterface) ([]*model.KVPair, string, error) {
	logContext := log.WithFields(log.Fields{
		"ListInterface": list,
		"Resource":      c.ResourceType,
		"Namespace":     c.Namespace,
	})
	logContext.Debug("List per-Node Resources")
	kvps := []*model.KVPair{}
	nodeName, resName, err := c.Converter.ListInterfaceToNodeAndName(list)
	if err != nil {
		logContext.WithError(err).Info("Failed to list resources: error in list interface conversion")
		return nil, "", err
	}
	ctx := context.Background()
	var nodes []apiv1.Node
	var rev string
	if nodeName != "" {
		newLogContext := logContext.WithField("NodeName", nodeName)
		node, err := c.ClientSet.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
		if err != nil {
			err = K8sErrorToCalico(err, nodeName)
			if _, ok := err.(errors.ErrorResourceDoesNotExist); !ok {
				newLogContext.WithError(err).Error("Failed to list resources: unable to query node")
				return nil, "", err
			}
			newLogContext.WithError(err).Warning("Return no results for resource list: node does not exist")
			return kvps, "", nil
		}
		nodes = append(nodes, *node)
		rev = node.GetResourceVersion()
	} else {
		nodeList, err := c.ClientSet.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
		if err != nil {
			logContext.WithError(err).Info("Failed to list resources: unable to list Nodes")
			return nil, "", K8sErrorToCalico(err, nodeName)
		}
		nodes = nodeList.Items
		rev = nodeList.GetResourceVersion()
	}
	for _, node := range nodes {
		nodeKVPs, err := c.extractResourcesFromAnnotation(&node, resName)
		if err != nil {
			logContext.WithField("NodeName", node.GetName()).WithError(err).Error("Error listing resources: error in data")
		}
		kvps = append(kvps, nodeKVPs...)
	}
	return kvps, rev, nil
}
func (c *customK8sNodeResourceClient) getNameAndNodeFromKey(key model.Key) (string, *apiv1.Node, error) {
	logContext := log.WithFields(log.Fields{
		"Key":       key,
		"Resource":  c.ResourceType,
		"Namespace": c.Namespace,
	})
	logContext.Debug("Extract node and resource info from Key")
	nodeName, resName, err := c.Converter.KeyToNodeAndName(key)
	if err != nil {
		logContext.WithError(err).Info("Error converting Key to Node and Resource name.")
		return "", nil, err
	}
	node, err := c.ClientSet.CoreV1().Nodes().Get(context.Background(), nodeName, metav1.GetOptions{})
	if err != nil {
		logContext.WithError(err).Info("Error getting Node configuration")
		return "", nil, K8sErrorToCalico(err, key)
	}
	return resName, node, nil
}
func (c *customK8sNodeResourceClient) nameToAnnotationKey(name string) string {
	return c.annotationKeyPrefix + name
}
func (c *customK8sNodeResourceClient) annotationKeyToName(key string) string {
	if strings.HasPrefix(key, c.annotationKeyPrefix) {
		return key[len(c.annotationKeyPrefix):]
	}
	return ""
}
func (c *customK8sNodeResourceClient) applyResourceToAnnotation(node *apiv1.Node, resName string, kvp *model.KVPair, action action) (*model.KVPair, error) {
	logContext := log.WithFields(log.Fields{
		"Value":     kvp.Value,
		"Resource":  c.ResourceType,
		"Action":    action,
		"Namespace": c.Namespace,
	})
	logContext.Debug("Updating value in annotation")
	data, err := model.SerializeValue(kvp)
	if err != nil {
		logContext.Error("Unable to convert value for annotation")
		return nil, err
	}
	if node.Annotations == nil {
		node.Annotations = map[string]string{}
	}
	node.Annotations[c.nameToAnnotationKey(resName)] = string(data)
	node, err = c.ClientSet.CoreV1().Nodes().Update(context.Background(), node, metav1.UpdateOptions{})
	if err != nil {
		logContext.WithError(err).Warning("Error updating Kubernetes Node")
		err = K8sErrorToCalico(err, kvp.Key)
		if _, ok := err.(errors.ErrorResourceUpdateConflict); ok {
			err = retryError{err: err}
		}
		return nil, err
	}
	return &model.KVPair{
		Key:      kvp.Key,
		Value:    kvp.Value,
		Revision: node.GetObjectMeta().GetResourceVersion(),
	}, nil
}