<commit_before>// Copyright 2017 Cisco Systems, Inc.
//
// 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 hostagent

import (
	"errors"
	"os"

	"github.com/sirupsen/logrus"

	"k8s.io/client-go/kubernetes"
	restclient "k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/clientcmd"

	netClientSet "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/client/clientset/versioned"
	md "github.com/noironetworks/aci-containers/pkg/metadata"
	nodeinfoclientset "github.com/noironetworks/aci-containers/pkg/nodeinfo/clientset/versioned"
	qospolicyclset "github.com/noironetworks/aci-containers/pkg/qospolicy/clientset/versioned"
	rdconfigclset "github.com/noironetworks/aci-containers/pkg/rdconfig/clientset/versioned"
	snatglobalclset "github.com/noironetworks/aci-containers/pkg/snatglobalinfo/clientset/versioned"
	snatlocalinfoclset "github.com/noironetworks/aci-containers/pkg/snatlocalinfo/clientset/versioned"
	snatpolicyclset "github.com/noironetworks/aci-containers/pkg/snatpolicy/clientset/versioned"
)

type Environment interface {
	Init(agent *HostAgent) error
	PrepareRun(stopCh <-chan struct{}) (bool, error)

	CniDeviceChanged(metadataKey *string, id *md.ContainerId)
	CniDeviceDeleted(metadataKey *string, id *md.ContainerId)

	CheckPodExists(metadataKey *string) (bool, error)
}

type K8sEnvironment struct {
	kubeClient          *kubernetes.Clientset
	snatGlobalClient    *snatglobalclset.Clientset
	snatPolicyClient    *snatpolicyclset.Clientset
	qosPolicyClient     *qospolicyclset.Clientset
	nodeInfo            *nodeinfoclientset.Clientset
	rdConfig            *rdconfigclset.Clientset
	snatLocalInfoClient *snatlocalinfoclset.Clientset
	agent               *HostAgent
	podInformer         cache.SharedIndexInformer
	endpointsInformer   cache.SharedIndexInformer
	serviceInformer     cache.SharedIndexInformer
	nodeInformer        cache.SharedIndexInformer
	netClient           *netClientSet.Clientset
}

func NewK8sEnvironment(config *HostAgentConfig, log *logrus.Logger) (*K8sEnvironment, error) {
	if config.NodeName == "" {
		config.NodeName = os.Getenv("KUBERNETES_NODE_NAME")
	}
	if config.NodeName == "" {
		err := errors.New("Node name not specified and $KUBERNETES_NODE_NAME empty")
		log.Error(err.Error())
		return nil, err
	}

	log.WithFields(logrus.Fields{
		"kubeconfig": config.KubeConfig,
		"node-name":  config.NodeName,
	}).Info("Setting up Kubernetes environment")

	log.Debug("Initializing kubernetes client")
	var restconfig *restclient.Config
	var err error
	if config.KubeConfig != "" {
		// use kubeconfig file from command line
		restconfig, err =
			clientcmd.BuildConfigFromFlags("", config.KubeConfig)
		if err != nil {
			return nil, err
		}
	} else {
		// creates the in-cluster config
		restconfig, err = restclient.InClusterConfig()
		if err != nil {
			return nil, err
		}
	}

	// creates the kubernetes API client
	kubeClient, err := kubernetes.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize kube client")
		return nil, err
	}
	snatGlobalClient, err := snatglobalclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snat global info client")
		return nil, err
	}
	nodeInfo, err := nodeinfoclientset.NewForConfig(restconfig)
	log.Debug("Initializing kubernetes client", nodeInfo)
	if err != nil {
		log.Debug("Failed to intialize node info client")
		return nil, err
	}
	snatPolicyClient, err := snatpolicyclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snatpolicy info client")
		return nil, err
	}
	qosPolicyClient, err := qospolicyclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snatpolicy info client")
		return nil, err
	}
	rdConfig, err := rdconfigclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snatpolicy info client")
		return nil, err
	}
	snatLocalInfoClient, err := snatlocalinfoclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snatpolicy info client")
		return nil, err
	}
	netClient, err := netClientSet.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize network attachment definition info client")
		return nil, err
	}

	return &K8sEnvironment{kubeClient: kubeClient, snatGlobalClient: snatGlobalClient,
		nodeInfo: nodeInfo, snatPolicyClient: snatPolicyClient, qosPolicyClient: qosPolicyClient, rdConfig: rdConfig, snatLocalInfoClient: snatLocalInfoClient, netClient: netClient}, nil
}

func (env *K8sEnvironment) Init(agent *HostAgent) error {
	env.agent = agent

	env.agent.log.Debug("Initializing informers")
	env.agent.initNodeInformerFromClient(env.kubeClient)
	env.agent.initPodInformerFromClient(env.kubeClient)
	env.agent.serviceEndPoints.InitClientInformer(env.kubeClient)
	env.agent.initServiceInformerFromClient(env.kubeClient)
	env.agent.initNamespaceInformerFromClient(env.kubeClient)
	env.agent.initNetworkPolicyInformerFromClient(env.kubeClient)
	env.agent.initDeploymentInformerFromClient(env.kubeClient)
	env.agent.initRCInformerFromClient(env.kubeClient)
	env.agent.initSnatGlobalInformerFromClient(env.snatGlobalClient)
	env.agent.initSnatPolicyInformerFromClient(env.snatPolicyClient)
	env.agent.initQoSPolicyInformerFromClient(env.qosPolicyClient)
	env.agent.initRdConfigInformerFromClient(env.rdConfig)
	env.agent.initQoSPolPodIndex()
	env.agent.initNetPolPodIndex()
	env.agent.initDepPodIndex()
	env.agent.initRCPodIndex()
	env.agent.initEventPoster(env.kubeClient)
	env.agent.initNetworkAttDefInformerFromClient(env.netClient)
	return nil
}

func (env *K8sEnvironment) PrepareRun(stopCh <-chan struct{}) (bool, error) {
	env.agent.log.Debug("Discovering node configuration")
	env.agent.updateOpflexConfig()
	go env.agent.runTickers(stopCh)

	env.agent.log.Debug("Starting node informer")
	go env.agent.nodeInformer.Run(stopCh)
	env.agent.log.Info("Waiting for node cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.nodeInformer.HasSynced)
	env.agent.log.Info("Node cache sync successful")

	env.agent.log.Debug("Starting service informer")
	go env.agent.serviceInformer.Run(stopCh)
	env.agent.log.Info("Waiting for service cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.serviceInformer.HasSynced)
	env.agent.log.Info("Service cache sync successful")

	env.agent.log.Debug("Starting snat global informer")
	go env.agent.snatGlobalInformer.Run(stopCh)
	env.agent.log.Info("Waiting for snat global cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.snatGlobalInformer.HasSynced)
	env.agent.log.Info("Snat global cache sync successful")

	env.agent.log.Debug("Starting snat policy informer")
	go env.agent.snatPolicyInformer.Run(stopCh)
	env.agent.log.Info("Waiting for snat policy sync")
	cache.WaitForCacheSync(stopCh, env.agent.snatPolicyInformer.HasSynced)
	env.agent.log.Info("Snat policy sync successful")

	env.agent.log.Debug("Starting rdConfig informer")
	go env.agent.rdConfigInformer.Run(stopCh)
	env.agent.log.Info("Waiting for rdConfig cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.rdConfigInformer.HasSynced)
	env.agent.log.Info("RdConfig cache sync successful")

	env.agent.log.Debug("Starting remaining informers")
	env.agent.log.Debug("Exporting node info: ", env.agent.config.NodeName)
	go env.agent.podInformer.Run(stopCh)
	cache.WaitForCacheSync(stopCh, env.agent.podInformer.HasSynced)
	env.agent.log.Info("Pod cache sync successful")
	go env.agent.controllerInformer.Run(stopCh)
	env.agent.serviceEndPoints.Run(stopCh)
	go env.agent.nsInformer.Run(stopCh)
	go env.agent.netPolInformer.Run(stopCh)
	go env.agent.depInformer.Run(stopCh)
	go env.agent.rcInformer.Run(stopCh)
	go env.agent.qosPolicyInformer.Run(stopCh)
	go env.agent.netAttDefInformer.Run(stopCh)
	cache.WaitForCacheSync(stopCh, env.agent.netAttDefInformer.HasSynced)
	env.agent.log.Info("Cache sync successful")
	return true, nil
}

func (env *K8sEnvironment) CniDeviceChanged(metadataKey *string, id *md.ContainerId) {
	env.agent.podChanged(metadataKey)
}

func (env *K8sEnvironment) CniDeviceDeleted(metadataKey *string, id *md.ContainerId) {
}

func (env *K8sEnvironment) CheckPodExists(metadataKey *string) (bool, error) {
	_, exists, err := env.agent.podInformer.GetStore().GetByKey(*metadataKey)
	return exists, err
}<commit_msg>add WaitForCacheSync on various informers<commit_after>// Copyright 2017 Cisco Systems, Inc.
//
// 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 hostagent

import (
	"errors"
	"os"

	"github.com/sirupsen/logrus"

	"k8s.io/client-go/kubernetes"
	restclient "k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/clientcmd"

	netClientSet "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/client/clientset/versioned"
	md "github.com/noironetworks/aci-containers/pkg/metadata"
	nodeinfoclientset "github.com/noironetworks/aci-containers/pkg/nodeinfo/clientset/versioned"
	qospolicyclset "github.com/noironetworks/aci-containers/pkg/qospolicy/clientset/versioned"
	rdconfigclset "github.com/noironetworks/aci-containers/pkg/rdconfig/clientset/versioned"
	snatglobalclset "github.com/noironetworks/aci-containers/pkg/snatglobalinfo/clientset/versioned"
	snatlocalinfoclset "github.com/noironetworks/aci-containers/pkg/snatlocalinfo/clientset/versioned"
	snatpolicyclset "github.com/noironetworks/aci-containers/pkg/snatpolicy/clientset/versioned"
)

type Environment interface {
	Init(agent *HostAgent) error
	PrepareRun(stopCh <-chan struct{}) (bool, error)

	CniDeviceChanged(metadataKey *string, id *md.ContainerId)
	CniDeviceDeleted(metadataKey *string, id *md.ContainerId)

	CheckPodExists(metadataKey *string) (bool, error)
}

type K8sEnvironment struct {
	kubeClient          *kubernetes.Clientset
	snatGlobalClient    *snatglobalclset.Clientset
	snatPolicyClient    *snatpolicyclset.Clientset
	qosPolicyClient     *qospolicyclset.Clientset
	nodeInfo            *nodeinfoclientset.Clientset
	rdConfig            *rdconfigclset.Clientset
	snatLocalInfoClient *snatlocalinfoclset.Clientset
	agent               *HostAgent
	podInformer         cache.SharedIndexInformer
	endpointsInformer   cache.SharedIndexInformer
	serviceInformer     cache.SharedIndexInformer
	nodeInformer        cache.SharedIndexInformer
	netClient           *netClientSet.Clientset
}

func NewK8sEnvironment(config *HostAgentConfig, log *logrus.Logger) (*K8sEnvironment, error) {
	if config.NodeName == "" {
		config.NodeName = os.Getenv("KUBERNETES_NODE_NAME")
	}
	if config.NodeName == "" {
		err := errors.New("Node name not specified and $KUBERNETES_NODE_NAME empty")
		log.Error(err.Error())
		return nil, err
	}

	log.WithFields(logrus.Fields{
		"kubeconfig": config.KubeConfig,
		"node-name":  config.NodeName,
	}).Info("Setting up Kubernetes environment")

	log.Debug("Initializing kubernetes client")
	var restconfig *restclient.Config
	var err error
	if config.KubeConfig != "" {
		// use kubeconfig file from command line
		restconfig, err =
			clientcmd.BuildConfigFromFlags("", config.KubeConfig)
		if err != nil {
			return nil, err
		}
	} else {
		// creates the in-cluster config
		restconfig, err = restclient.InClusterConfig()
		if err != nil {
			return nil, err
		}
	}

	// creates the kubernetes API client
	kubeClient, err := kubernetes.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize kube client")
		return nil, err
	}
	snatGlobalClient, err := snatglobalclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snat global info client")
		return nil, err
	}
	nodeInfo, err := nodeinfoclientset.NewForConfig(restconfig)
	log.Debug("Initializing kubernetes client", nodeInfo)
	if err != nil {
		log.Debug("Failed to intialize node info client")
		return nil, err
	}
	snatPolicyClient, err := snatpolicyclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snatpolicy info client")
		return nil, err
	}
	qosPolicyClient, err := qospolicyclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snatpolicy info client")
		return nil, err
	}
	rdConfig, err := rdconfigclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snatpolicy info client")
		return nil, err
	}
	snatLocalInfoClient, err := snatlocalinfoclset.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize snatpolicy info client")
		return nil, err
	}
	netClient, err := netClientSet.NewForConfig(restconfig)
	if err != nil {
		log.Debug("Failed to intialize network attachment definition info client")
		return nil, err
	}

	return &K8sEnvironment{kubeClient: kubeClient, snatGlobalClient: snatGlobalClient,
		nodeInfo: nodeInfo, snatPolicyClient: snatPolicyClient, qosPolicyClient: qosPolicyClient, rdConfig: rdConfig, snatLocalInfoClient: snatLocalInfoClient, netClient: netClient}, nil
}

func (env *K8sEnvironment) Init(agent *HostAgent) error {
	env.agent = agent

	env.agent.log.Debug("Initializing informers")
	env.agent.initNodeInformerFromClient(env.kubeClient)
	env.agent.initPodInformerFromClient(env.kubeClient)
	env.agent.serviceEndPoints.InitClientInformer(env.kubeClient)
	env.agent.initServiceInformerFromClient(env.kubeClient)
	env.agent.initNamespaceInformerFromClient(env.kubeClient)
	env.agent.initNetworkPolicyInformerFromClient(env.kubeClient)
	env.agent.initDeploymentInformerFromClient(env.kubeClient)
	env.agent.initRCInformerFromClient(env.kubeClient)
	env.agent.initSnatGlobalInformerFromClient(env.snatGlobalClient)
	env.agent.initSnatPolicyInformerFromClient(env.snatPolicyClient)
	env.agent.initQoSPolicyInformerFromClient(env.qosPolicyClient)
	env.agent.initRdConfigInformerFromClient(env.rdConfig)
	env.agent.initQoSPolPodIndex()
	env.agent.initNetPolPodIndex()
	env.agent.initDepPodIndex()
	env.agent.initRCPodIndex()
	env.agent.initEventPoster(env.kubeClient)
	env.agent.initNetworkAttDefInformerFromClient(env.netClient)
	return nil
}

func (env *K8sEnvironment) PrepareRun(stopCh <-chan struct{}) (bool, error) {
	env.agent.log.Debug("Discovering node configuration")
	env.agent.updateOpflexConfig()
	go env.agent.runTickers(stopCh)

	env.agent.log.Debug("Starting node informer")
	go env.agent.nodeInformer.Run(stopCh)
	env.agent.log.Info("Waiting for node cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.nodeInformer.HasSynced)
	env.agent.log.Info("Node cache sync successful")

	env.agent.log.Debug("Starting service informer")
	go env.agent.serviceInformer.Run(stopCh)
	env.agent.log.Info("Waiting for service cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.serviceInformer.HasSynced)
	env.agent.log.Info("Service cache sync successful")

	env.agent.log.Debug("Starting snat global informer")
	go env.agent.snatGlobalInformer.Run(stopCh)
	env.agent.log.Info("Waiting for snat global cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.snatGlobalInformer.HasSynced)
	env.agent.log.Info("Snat global cache sync successful")

	env.agent.log.Debug("Starting snat policy informer")
	go env.agent.snatPolicyInformer.Run(stopCh)
	env.agent.log.Info("Waiting for snat policy sync")
	cache.WaitForCacheSync(stopCh, env.agent.snatPolicyInformer.HasSynced)
	env.agent.log.Info("Snat policy sync successful")

	env.agent.log.Debug("Starting rdConfig informer")
	go env.agent.rdConfigInformer.Run(stopCh)
	env.agent.log.Info("Waiting for rdConfig cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.rdConfigInformer.HasSynced)
	env.agent.log.Info("RdConfig cache sync successful")

	env.agent.log.Debug("Starting remaining informers")
	env.agent.log.Debug("Exporting node info: ", env.agent.config.NodeName)
	go env.agent.podInformer.Run(stopCh)
	cache.WaitForCacheSync(stopCh, env.agent.podInformer.HasSynced)
	env.agent.log.Info("Pod cache sync successful")

	env.agent.log.Debug("Starting controller informers")
	go env.agent.controllerInformer.Run(stopCh)
	env.agent.log.Info("Waiting for controller cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.controllerInformer.HasSynced)
	env.agent.log.Info("controller cache sync successful")

	env.agent.serviceEndPoints.Run(stopCh)

	env.agent.log.Debug("Starting namespace informers")
	go env.agent.nsInformer.Run(stopCh)
	env.agent.log.Info("Waiting for namespace cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.nsInformer.HasSynced)
	env.agent.log.Info("namespace cache sync successful")

	env.agent.log.Debug("Starting networkPolicy informers")
	go env.agent.netPolInformer.Run(stopCh)
	env.agent.log.Info("Waiting for networkPolicy cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.netPolInformer.HasSynced)
	env.agent.log.Info("networkPolicy cache sync successful")

	env.agent.log.Debug("Starting deployment informers")
	go env.agent.depInformer.Run(stopCh)
	env.agent.log.Info("Waiting for deployment cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.depInformer.HasSynced)
	env.agent.log.Info("deployment cache sync successful")

	env.agent.log.Debug("Starting ReplicationController informers")
	go env.agent.rcInformer.Run(stopCh)
	env.agent.log.Info("Waiting for ReplicationController cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.rcInformer.HasSynced)
	env.agent.log.Info("ReplicationController cache sync successful")

	env.agent.log.Debug("Starting qosPolicy informers")
	go env.agent.qosPolicyInformer.Run(stopCh)
	env.agent.log.Info("Waiting for qosPolicy cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.qosPolicyInformer.HasSynced)
	env.agent.log.Info("qosPolicy cache sync successful")

	env.agent.log.Debug("Starting netAttDef informers")
	go env.agent.netAttDefInformer.Run(stopCh)
	env.agent.log.Info("Waiting for netAttDef cache sync")
	cache.WaitForCacheSync(stopCh, env.agent.netAttDefInformer.HasSynced)
	env.agent.log.Info("netAttDef cache sync successful")

	env.agent.log.Info("Cache sync successful")
	return true, nil
}

func (env *K8sEnvironment) CniDeviceChanged(metadataKey *string, id *md.ContainerId) {
	env.agent.podChanged(metadataKey)
}

func (env *K8sEnvironment) CniDeviceDeleted(metadataKey *string, id *md.ContainerId) {
}

func (env *K8sEnvironment) CheckPodExists(metadataKey *string) (bool, error) {
	_, exists, err := env.agent.podInformer.GetStore().GetByKey(*metadataKey)
	return exists, err
}
{'fraction_non_alphanumeric': 0.10737104565996572, 'fraction_numerical': 0.0022855955534777416, 'mean_word_length': 5.430193720774883, 'pattern_counts': {'":': 4, '<': 7, '<?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': '5081865', 'n_tokens_mistral': 6561, 'n_tokens_neox': 6399, 'n_words': 1262}