<commit_before>// Copyright (c) 2016-2017 Tigera, Inc. All rights reserved.

// 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 (
	goerrors "errors"
	"fmt"
	"strings"
	"time"

	log "github.com/Sirupsen/logrus"

	"github.com/projectcalico/libcalico-go/lib/backend/api"
	"github.com/projectcalico/libcalico-go/lib/backend/k8s/resources"
	"github.com/projectcalico/libcalico-go/lib/backend/k8s/thirdparty"
	"github.com/projectcalico/libcalico-go/lib/backend/model"
	"github.com/projectcalico/libcalico-go/lib/errors"

	"k8s.io/client-go/kubernetes"
	clientapi "k8s.io/client-go/pkg/api"
	kerrors "k8s.io/client-go/pkg/api/errors"
	kapiv1 "k8s.io/client-go/pkg/api/v1"
	extensions "k8s.io/client-go/pkg/apis/extensions/v1beta1"
	metav1 "k8s.io/client-go/pkg/apis/meta/v1"
	"k8s.io/client-go/pkg/runtime"
	"k8s.io/client-go/pkg/runtime/schema"
	"k8s.io/client-go/pkg/runtime/serializer"
	"k8s.io/client-go/pkg/util/wait"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"github.com/projectcalico/libcalico-go/lib/net"
	v1 "k8s.io/client-go/pkg/api/v1"
)

type KubeClient struct {
	// Main Kubernetes clients.
	clientSet *kubernetes.Clientset

	// Client for interacting with ThirdPartyResources.
	tprClient *rest.RESTClient

	// Contains methods for converting Kubernetes resources to
	// Calico resources.
	converter converter

	// Clients for interacting with Calico resources.
	ipPoolClient api.Client
	nodeClient   api.Client
}

type KubeConfig struct {
	Kubeconfig               string `json:"kubeconfig" envconfig:"KUBECONFIG" default:""`
	K8sAPIEndpoint           string `json:"k8sAPIEndpoint" envconfig:"K8S_API_ENDPOINT" default:""`
	K8sKeyFile               string `json:"k8sKeyFile" envconfig:"K8S_KEY_FILE" default:""`
	K8sCertFile              string `json:"k8sCertFile" envconfig:"K8S_CERT_FILE" default:""`
	K8sCAFile                string `json:"k8sCAFile" envconfig:"K8S_CA_FILE" default:""`
	K8sAPIToken              string `json:"k8sAPIToken" envconfig:"K8S_API_TOKEN" default:""`
	K8sInsecureSkipTLSVerify bool   `json:"k8sInsecureSkipTLSVerify" envconfig:"K8S_INSECURE_SKIP_TLS_VERIFY" default:""`
}

func NewKubeClient(kc *KubeConfig) (*KubeClient, error) {
	// Use the kubernetes client code to load the kubeconfig file and combine it with the overrides.
	log.Debugf("Building client for config: %+v", kc)
	configOverrides := &clientcmd.ConfigOverrides{}
	var overridesMap = []struct {
		variable *string
		value    string
	}{
		{&configOverrides.ClusterInfo.Server, kc.K8sAPIEndpoint},
		{&configOverrides.AuthInfo.ClientCertificate, kc.K8sCertFile},
		{&configOverrides.AuthInfo.ClientKey, kc.K8sKeyFile},
		{&configOverrides.ClusterInfo.CertificateAuthority, kc.K8sCAFile},
		{&configOverrides.AuthInfo.Token, kc.K8sAPIToken},
	}

	// Set an explicit path to the kubeconfig if one
	// was provided.
	loadingRules := clientcmd.ClientConfigLoadingRules{}
	if kc.Kubeconfig != "" {
		loadingRules.ExplicitPath = kc.Kubeconfig
	}

	// Using the override map above, populate any non-empty values.
	for _, override := range overridesMap {
		if override.value != "" {
			*override.variable = override.value
		}
	}
	if kc.K8sInsecureSkipTLSVerify {
		configOverrides.ClusterInfo.InsecureSkipTLSVerify = true
	}
	log.Debugf("Config overrides: %+v", configOverrides)

	// A kubeconfig file was provided.  Use it to load a config, passing through
	// any overrides.
	config, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
		&loadingRules, configOverrides).ClientConfig()
	if err != nil {
		return nil, resources.K8sErrorToCalico(err, nil)
	}

	// Create the clientset
	cs, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, resources.K8sErrorToCalico(err, nil)
	}
	log.Debugf("Created k8s clientSet: %+v", cs)

	tprClient, err := buildTPRClient(config)
	if err != nil {
		return nil, fmt.Errorf("Failed to build TPR client: %s", err)
	}
	kubeClient := &KubeClient{
		clientSet: cs,
		tprClient: tprClient,
	}

	// Create the Calico sub-clients.
	kubeClient.ipPoolClient = resources.NewIPPools(cs, tprClient)
	kubeClient.nodeClient = resources.NewNodeClient(cs, tprClient)

	return kubeClient, nil
}

func (c *KubeClient) EnsureInitialized() error {
	// Ensure the necessary ThirdPartyResources exist in the API.
	log.Info("Ensuring ThirdPartyResources exist")
	err := c.ensureThirdPartyResources()
	if err != nil {
		return fmt.Errorf("Failed to ensure ThirdPartyResources exist: %s", err)
	}
	log.Info("ThirdPartyResources exist")

	// Ensure ClusterType is set.
	log.Info("Ensuring ClusterType is set")
	err = c.waitForClusterType()
	if err != nil {
		return fmt.Errorf("Failed to ensure ClusterType is set: %s", err)
	}
	log.Info("ClusterType is set")
	return nil
}

func (c *KubeClient) EnsureCalicoNodeInitialized(node string) error {
	log.WithField("Node", node).Info("Ensuring node is initialized")
	return nil
}

// ensureThirdPartyResources ensures the necessary thirdparty resources are created
// and will retry every second for 30 seconds or until they exist.
func (c *KubeClient) ensureThirdPartyResources() error {
	return wait.PollImmediate(1*time.Second, 30*time.Second, func() (bool, error) {
		if err := c.createThirdPartyResources(); err != nil {
			return false, err
		}
		return true, nil
	})
}

// createThirdPartyResources creates the necessary third party resources if they
// do not already exist.
func (c *KubeClient) createThirdPartyResources() error {
	// Ensure a resource exists for Calico global configuration.
	log.Info("Ensuring GlobalConfig ThirdPartyResource exists")
	tpr := extensions.ThirdPartyResource{
		ObjectMeta: kapiv1.ObjectMeta{
			Name:      "global-config.projectcalico.org",
			Namespace: "kube-system",
		},
		Description: "Calico Global Configuration",
		Versions:    []extensions.APIVersion{{Name: "v1"}},
	}
	_, err := c.clientSet.Extensions().ThirdPartyResources().Create(&tpr)
	if err != nil {
		// Don't care if it already exists.
		if !kerrors.IsAlreadyExists(err) {
			return resources.K8sErrorToCalico(err, tpr)
		}
	}

	// Ensure the IP Pool TPR exists.
	return c.ipPoolClient.EnsureInitialized()
}

// waitForClusterType polls until GlobalConfig is ready, or until 30 seconds have passed.
func (c *KubeClient) waitForClusterType() error {
	return wait.PollImmediate(1*time.Second, 30*time.Second, func() (bool, error) {
		return c.ensureClusterType()
	})
}

// ensureClusterType ensures that the ClusterType is configured.
func (c *KubeClient) ensureClusterType() (bool, error) {
	k := model.GlobalConfigKey{
		Name: "ClusterType",
	}
	value := "KDD"

	// See if a cluster type has been set.  If so, append
	// any existing values to it.
	ct, err := c.Get(k)
	if err != nil {
		if _, ok := err.(errors.ErrorResourceDoesNotExist); !ok {
			// Resource exists but we got another error.
			return false, err
		}
		// Resource does not exist.
	}
	if ct != nil {
		existingValue := ct.Value.(string)
		if !strings.Contains(existingValue, "KDD") {
			existingValue = fmt.Sprintf("%s,KDD", existingValue)
		}
		value = existingValue
	}
	log.WithField("value", value).Debug("Setting ClusterType")
	_, err = c.Apply(&model.KVPair{
		Key:   k,
		Value: value,
	})
	if err != nil {
		// Don't return an error, but indicate that we need
		// to retry.
		log.Warnf("Failed to apply ClusterType: %s", err)
		return false, nil
	}
	return true, nil
}

// buildTPRClient builds a RESTClient configured to interact with Calico ThirdPartyResources
func buildTPRClient(baseConfig *rest.Config) (*rest.RESTClient, error) {
	// Generate config using the base config.
	cfg := baseConfig
	cfg.GroupVersion = &schema.GroupVersion{
		Group:   "projectcalico.org",
		Version: "v1",
	}
	cfg.APIPath = "/apis"
	cfg.ContentType = runtime.ContentTypeJSON
	cfg.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: clientapi.Codecs}

	cli, err := rest.RESTClientFor(cfg)
	if err != nil {
		return nil, err
	}

	// We also need to register resources.
	schemeBuilder := runtime.NewSchemeBuilder(
		func(scheme *runtime.Scheme) error {
			scheme.AddKnownTypes(
				*cfg.GroupVersion,
				&thirdparty.GlobalConfig{},
				&thirdparty.GlobalConfigList{},
				&thirdparty.IpPool{},
				&thirdparty.IpPoolList{},
				&kapiv1.ListOptions{},
				&kapiv1.DeleteOptions{},
			)
			return nil
		})
	schemeBuilder.AddToScheme(clientapi.Scheme)

	return cli, nil
}

func (c *KubeClient) Syncer(callbacks api.SyncerCallbacks) api.Syncer {
	return newSyncer(*c, callbacks)
}

// Create an entry in the datastore.  This errors if the entry already exists.
func (c *KubeClient) Create(d *model.KVPair) (*model.KVPair, error) {
	log.Debugf("Performing 'Create' for %+v", d)
	switch d.Key.(type) {
	case model.GlobalConfigKey:
		return c.createGlobalConfig(d)
	case model.IPPoolKey:
		return c.ipPoolClient.Create(d)
	case model.NodeKey:
		return c.nodeClient.Create(d)
	default:
		log.Warn("Attempt to 'Create' using kubernetes backend is not supported.")
		return nil, errors.ErrorOperationNotSupported{
			Identifier: d.Key,
			Operation:  "Create",
		}
	}
}

// Update an existing entry in the datastore.  This errors if the entry does
// not exist.
func (c *KubeClient) Update(d *model.KVPair) (*model.KVPair, error) {
	log.Debugf("Performing 'Update' for %+v", d)
	switch d.Key.(type) {
	case model.GlobalConfigKey:
		return c.updateGlobalConfig(d)
	case model.IPPoolKey:
		return c.ipPoolClient.Update(d)
	case model.NodeKey:
		return c.nodeClient.Update(d)
	default:
		// If the resource isn't supported, then this is a no-op.
		log.Debugf("'Update' for %+v is no-op", d.Key)
		return d, nil
	}
}

// Set an existing entry in the datastore.  This ignores whether an entry already
// exists.
func (c *KubeClient) Apply(d *model.KVPair) (*model.KVPair, error) {
	log.Debugf("Performing 'Apply' for %+v", d)
	switch d.Key.(type) {
	case model.WorkloadEndpointKey:
		return c.applyWorkloadEndpoint(d)
	case model.GlobalConfigKey:
		return c.applyGlobalConfig(d)
	case model.IPPoolKey:
		return c.ipPoolClient.Apply(d)
	case model.NodeKey:
		return c.nodeClient.Apply(d)
	default:
		log.Debugf("'Apply' for %s is no-op", d.Key)
		return d, nil
	}
}

// Delete an entry in the datastore. This is a no-op when using the k8s backend.
func (c *KubeClient) Delete(d *model.KVPair) error {
	log.Debugf("Performing 'Delete' for %+v", d)
	switch d.Key.(type) {
	case model.GlobalConfigKey:
		return c.deleteGlobalConfig(d)
	case model.IPPoolKey:
		return c.ipPoolClient.Delete(d)
	case model.NodeKey:
		return c.nodeClient.Delete(d)
	default:
		log.Warn("Attempt to 'Delete' using kubernetes datastore driver is not supported.")
		return nil
	}
}

// Get an entry from the datastore.  This errors if the entry does not exist.<commit_msg>KDD raise a not supported error when an action is not supported for a resource type.<commit_after>// Copyright (c) 2016-2017 Tigera, Inc. All rights reserved.

// 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 (
	goerrors "errors"
	"fmt"
	"strings"
	"time"

	log "github.com/Sirupsen/logrus"

	"github.com/projectcalico/libcalico-go/lib/backend/api"
	"github.com/projectcalico/libcalico-go/lib/backend/k8s/resources"
	"github.com/projectcalico/libcalico-go/lib/backend/k8s/thirdparty"
	"github.com/projectcalico/libcalico-go/lib/backend/model"
	"github.com/projectcalico/libcalico-go/lib/errors"

	"k8s.io/client-go/kubernetes"
	clientapi "k8s.io/client-go/pkg/api"
	kerrors "k8s.io/client-go/pkg/api/errors"
	kapiv1 "k8s.io/client-go/pkg/api/v1"
	extensions "k8s.io/client-go/pkg/apis/extensions/v1beta1"
	metav1 "k8s.io/client-go/pkg/apis/meta/v1"
	"k8s.io/client-go/pkg/runtime"
	"k8s.io/client-go/pkg/runtime/schema"
	"k8s.io/client-go/pkg/runtime/serializer"
	"k8s.io/client-go/pkg/util/wait"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"github.com/projectcalico/libcalico-go/lib/net"
	v1 "k8s.io/client-go/pkg/api/v1"
)

type KubeClient struct {
	// Main Kubernetes clients.
	clientSet *kubernetes.Clientset

	// Client for interacting with ThirdPartyResources.
	tprClient *rest.RESTClient

	// Contains methods for converting Kubernetes resources to
	// Calico resources.
	converter converter

	// Clients for interacting with Calico resources.
	ipPoolClient api.Client
	nodeClient   api.Client
}

type KubeConfig struct {
	Kubeconfig               string `json:"kubeconfig" envconfig:"KUBECONFIG" default:""`
	K8sAPIEndpoint           string `json:"k8sAPIEndpoint" envconfig:"K8S_API_ENDPOINT" default:""`
	K8sKeyFile               string `json:"k8sKeyFile" envconfig:"K8S_KEY_FILE" default:""`
	K8sCertFile              string `json:"k8sCertFile" envconfig:"K8S_CERT_FILE" default:""`
	K8sCAFile                string `json:"k8sCAFile" envconfig:"K8S_CA_FILE" default:""`
	K8sAPIToken              string `json:"k8sAPIToken" envconfig:"K8S_API_TOKEN" default:""`
	K8sInsecureSkipTLSVerify bool   `json:"k8sInsecureSkipTLSVerify" envconfig:"K8S_INSECURE_SKIP_TLS_VERIFY" default:""`
}

func NewKubeClient(kc *KubeConfig) (*KubeClient, error) {
	// Use the kubernetes client code to load the kubeconfig file and combine it with the overrides.
	log.Debugf("Building client for config: %+v", kc)
	configOverrides := &clientcmd.ConfigOverrides{}
	var overridesMap = []struct {
		variable *string
		value    string
	}{
		{&configOverrides.ClusterInfo.Server, kc.K8sAPIEndpoint},
		{&configOverrides.AuthInfo.ClientCertificate, kc.K8sCertFile},
		{&configOverrides.AuthInfo.ClientKey, kc.K8sKeyFile},
		{&configOverrides.ClusterInfo.CertificateAuthority, kc.K8sCAFile},
		{&configOverrides.AuthInfo.Token, kc.K8sAPIToken},
	}

	// Set an explicit path to the kubeconfig if one
	// was provided.
	loadingRules := clientcmd.ClientConfigLoadingRules{}
	if kc.Kubeconfig != "" {
		loadingRules.ExplicitPath = kc.Kubeconfig
	}

	// Using the override map above, populate any non-empty values.
	for _, override := range overridesMap {
		if override.value != "" {
			*override.variable = override.value
		}
	}
	if kc.K8sInsecureSkipTLSVerify {
		configOverrides.ClusterInfo.InsecureSkipTLSVerify = true
	}
	log.Debugf("Config overrides: %+v", configOverrides)

	// A kubeconfig file was provided.  Use it to load a config, passing through
	// any overrides.
	config, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
		&loadingRules, configOverrides).ClientConfig()
	if err != nil {
		return nil, resources.K8sErrorToCalico(err, nil)
	}

	// Create the clientset
	cs, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, resources.K8sErrorToCalico(err, nil)
	}
	log.Debugf("Created k8s clientSet: %+v", cs)

	tprClient, err := buildTPRClient(config)
	if err != nil {
		return nil, fmt.Errorf("Failed to build TPR client: %s", err)
	}
	kubeClient := &KubeClient{
		clientSet: cs,
		tprClient: tprClient,
	}

	// Create the Calico sub-clients.
	kubeClient.ipPoolClient = resources.NewIPPools(cs, tprClient)
	kubeClient.nodeClient = resources.NewNodeClient(cs, tprClient)

	return kubeClient, nil
}

func (c *KubeClient) EnsureInitialized() error {
	// Ensure the necessary ThirdPartyResources exist in the API.
	log.Info("Ensuring ThirdPartyResources exist")
	err := c.ensureThirdPartyResources()
	if err != nil {
		return fmt.Errorf("Failed to ensure ThirdPartyResources exist: %s", err)
	}
	log.Info("ThirdPartyResources exist")

	// Ensure ClusterType is set.
	log.Info("Ensuring ClusterType is set")
	err = c.waitForClusterType()
	if err != nil {
		return fmt.Errorf("Failed to ensure ClusterType is set: %s", err)
	}
	log.Info("ClusterType is set")
	return nil
}

func (c *KubeClient) EnsureCalicoNodeInitialized(node string) error {
	log.WithField("Node", node).Info("Ensuring node is initialized")
	return nil
}

// ensureThirdPartyResources ensures the necessary thirdparty resources are created
// and will retry every second for 30 seconds or until they exist.
func (c *KubeClient) ensureThirdPartyResources() error {
	return wait.PollImmediate(1*time.Second, 30*time.Second, func() (bool, error) {
		if err := c.createThirdPartyResources(); err != nil {
			return false, err
		}
		return true, nil
	})
}

// createThirdPartyResources creates the necessary third party resources if they
// do not already exist.
func (c *KubeClient) createThirdPartyResources() error {
	// Ensure a resource exists for Calico global configuration.
	log.Info("Ensuring GlobalConfig ThirdPartyResource exists")
	tpr := extensions.ThirdPartyResource{
		ObjectMeta: kapiv1.ObjectMeta{
			Name:      "global-config.projectcalico.org",
			Namespace: "kube-system",
		},
		Description: "Calico Global Configuration",
		Versions:    []extensions.APIVersion{{Name: "v1"}},
	}
	_, err := c.clientSet.Extensions().ThirdPartyResources().Create(&tpr)
	if err != nil {
		// Don't care if it already exists.
		if !kerrors.IsAlreadyExists(err) {
			return resources.K8sErrorToCalico(err, tpr)
		}
	}

	// Ensure the IP Pool TPR exists.
	return c.ipPoolClient.EnsureInitialized()
}

// waitForClusterType polls until GlobalConfig is ready, or until 30 seconds have passed.
func (c *KubeClient) waitForClusterType() error {
	return wait.PollImmediate(1*time.Second, 30*time.Second, func() (bool, error) {
		return c.ensureClusterType()
	})
}

// ensureClusterType ensures that the ClusterType is configured.
func (c *KubeClient) ensureClusterType() (bool, error) {
	k := model.GlobalConfigKey{
		Name: "ClusterType",
	}
	value := "KDD"

	// See if a cluster type has been set.  If so, append
	// any existing values to it.
	ct, err := c.Get(k)
	if err != nil {
		if _, ok := err.(errors.ErrorResourceDoesNotExist); !ok {
			// Resource exists but we got another error.
			return false, err
		}
		// Resource does not exist.
	}
	if ct != nil {
		existingValue := ct.Value.(string)
		if !strings.Contains(existingValue, "KDD") {
			existingValue = fmt.Sprintf("%s,KDD", existingValue)
		}
		value = existingValue
	}
	log.WithField("value", value).Debug("Setting ClusterType")
	_, err = c.Apply(&model.KVPair{
		Key:   k,
		Value: value,
	})
	if err != nil {
		// Don't return an error, but indicate that we need
		// to retry.
		log.Warnf("Failed to apply ClusterType: %s", err)
		return false, nil
	}
	return true, nil
}

// buildTPRClient builds a RESTClient configured to interact with Calico ThirdPartyResources
func buildTPRClient(baseConfig *rest.Config) (*rest.RESTClient, error) {
	// Generate config using the base config.
	cfg := baseConfig
	cfg.GroupVersion = &schema.GroupVersion{
		Group:   "projectcalico.org",
		Version: "v1",
	}
	cfg.APIPath = "/apis"
	cfg.ContentType = runtime.ContentTypeJSON
	cfg.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: clientapi.Codecs}

	cli, err := rest.RESTClientFor(cfg)
	if err != nil {
		return nil, err
	}

	// We also need to register resources.
	schemeBuilder := runtime.NewSchemeBuilder(
		func(scheme *runtime.Scheme) error {
			scheme.AddKnownTypes(
				*cfg.GroupVersion,
				&thirdparty.GlobalConfig{},
				&thirdparty.GlobalConfigList{},
				&thirdparty.IpPool{},
				&thirdparty.IpPoolList{},
				&kapiv1.ListOptions{},
				&kapiv1.DeleteOptions{},
			)
			return nil
		})
	schemeBuilder.AddToScheme(clientapi.Scheme)

	return cli, nil
}

func (c *KubeClient) Syncer(callbacks api.SyncerCallbacks) api.Syncer {
	return newSyncer(*c, callbacks)
}

// Create an entry in the datastore.  This errors if the entry already exists.
func (c *KubeClient) Create(d *model.KVPair) (*model.KVPair, error) {
	log.Debugf("Performing 'Create' for %+v", d)
	switch d.Key.(type) {
	case model.GlobalConfigKey:
		return c.createGlobalConfig(d)
	case model.IPPoolKey:
		return c.ipPoolClient.Create(d)
	case model.NodeKey:
		return c.nodeClient.Create(d)
	default:
		log.Warn("Attempt to 'Create' using kubernetes backend is not supported.")
		return nil, errors.ErrorOperationNotSupported{
			Identifier: d.Key,
			Operation:  "Create",
		}
	}
}

// Update an existing entry in the datastore.  This errors if the entry does
// not exist.
func (c *KubeClient) Update(d *model.KVPair) (*model.KVPair, error) {
	log.Debugf("Performing 'Update' for %+v", d)
	switch d.Key.(type) {
	case model.GlobalConfigKey:
		return c.updateGlobalConfig(d)
	case model.IPPoolKey:
		return c.ipPoolClient.Update(d)
	case model.NodeKey:
		return c.nodeClient.Update(d)
	default:
		log.Warn("Attempt to 'Update' using kubernetes backend is not supported.")
		return nil, errors.ErrorOperationNotSupported{
			Identifier: d.Key,
			Operation:  "Update",
		}
	}
}

// Set an existing entry in the datastore.  This ignores whether an entry already
// exists.
func (c *KubeClient) Apply(d *model.KVPair) (*model.KVPair, error) {
	log.Debugf("Performing 'Apply' for %+v", d)
	switch d.Key.(type) {
	case model.WorkloadEndpointKey:
		return c.applyWorkloadEndpoint(d)
	case model.GlobalConfigKey:
		return c.applyGlobalConfig(d)
	case model.IPPoolKey:
		return c.ipPoolClient.Apply(d)
	case model.NodeKey:
		return c.nodeClient.Apply(d)
	default:
		log.Warn("Attempt to 'Apply' using kubernetes backend is not supported.")
		return nil, errors.ErrorOperationNotSupported{
			Identifier: d.Key,
			Operation:  "Apply",
		}
	}
}

// Delete an entry in the datastore. This is a no-op when using the k8s backend.
func (c *KubeClient) Delete(d *model.KVPair) error {
	log.Debugf("Performing 'Delete' for %+v", d)
	switch d.Key.(type) {
	case model.GlobalConfigKey:
		return c.deleteGlobalConfig(d)
	case model.IPPoolKey:
		return c.ipPoolClient.Delete(d)
	case model.NodeKey:
		return c.nodeClient.Delete(d)
	default:
		log.Warn("Attempt to 'Delete' using kubernetes backend is not supported.")
		return errors.ErrorOperationNotSupported{
			Identifier: d.Key,
			Operation:  "Delete",
		}
	}
}

// Get an entry from the datastore.  This errors if the entry does not exist.
{'fraction_non_alphanumeric': 0.11861281826163302, 'fraction_numerical': 0.007023705004389816, 'mean_word_length': 4.176323562826631, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6777429', 'n_tokens_mistral': 8098, 'n_tokens_neox': 7716, 'n_words': 2009}