<commit_before>package controller

import (
	"context"
	"fmt"
	"time"

	clientset "github.com/matt-tyler/elasticsearch-operator/pkg/client/clientset/versioned"
	informers "github.com/matt-tyler/elasticsearch-operator/pkg/client/informers/externalversions"
	listers "github.com/matt-tyler/elasticsearch-operator/pkg/client/listers/es/v1"
	"github.com/matt-tyler/elasticsearch-operator/pkg/log"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/util/runtime"

	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	kubeinformers "k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	corelisters "k8s.io/client-go/listers/core/v1"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
)

const maxRetries = 5

type Controller struct {
	log.Logger

	kubeclientset kubernetes.Interface
	esclientset   clientset.Interface

	kubeInformerFactory kubeinformers.SharedInformerFactory
	esInformerFactory   informers.SharedInformerFactory

	clustersSynced cache.InformerSynced
	servicesSynced cache.InformerSynced

	clusterLister listers.ClusterLister
	serviceLister corelisters.ServiceLister

	queue workqueue.RateLimitingInterface
}

func NewController(config *rest.Config) *Controller {
	queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())

	kubeclientset := kubernetes.NewForConfigOrDie(config)
	esclientset := clientset.NewForConfigOrDie(config)

	resyncPeriod := 0 * time.Second

	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeclientset, resyncPeriod)
	esInformerFactory := informers.NewSharedInformerFactory(esclientset, resyncPeriod)

	clusterInformer := esInformerFactory.Es().V1().Clusters()
	serviceInformer := kubeInformerFactory.Core().V1().Services()

	clusterInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			if key, err := cache.MetaNamespaceKeyFunc(obj); err == nil {
				queue.Add(key)
			}
		},
		UpdateFunc: func(oldObj interface{}, newObj interface{}) {
			if key, err := cache.MetaNamespaceKeyFunc(newObj); err == nil {
				queue.Add(key)
			}
		},
		DeleteFunc: func(obj interface{}) {
			if key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj); err == nil {
				queue.Add(key)
			}
		},
	})

	ctx := context.Background()

	go kubeInformerFactory.Start(ctx.Done())
	go esInformerFactory.Start(ctx.Done())

	return &Controller{
		log.NewLogger(),
		kubeclientset,
		esclientset,
		kubeInformerFactory,
		esInformerFactory,
		clusterInformer.Informer().HasSynced,
		serviceInformer.Informer().HasSynced,
		clusterInformer.Lister(),
		serviceInformer.Lister(),
		queue,
	}
}

func (c *Controller) runWorker() {
	for c.processNextItem() {
		// continue looping
	}
}

func (c *Controller) processNextItem() bool {
	obj, shutdown := c.queue.Get()

	if shutdown {
		return false
	}

	err := func(obj interface{}) error {
		defer c.queue.Done(obj)

		var key string
		var ok bool

		if key, ok = obj.(string); !ok {
			c.queue.Forget(obj)
			runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
			return nil
		}

		if err := c.sync(key); err != nil {
			return fmt.Errorf("error syncing '%s': %s", key, err.Error())
		}

		c.queue.Forget(obj)
		return nil
	}(obj)

	if err != nil {
		runtime.HandleError(err)
	}

	return true
}

func (c *Controller) sync(key string) error {
	c.Infof("Processing change to %s", key)

	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		runtime.HandleError(fmt.Errorf("invalid resource key: %s", key))
		return nil
	}

	cluster, err := c.clusterLister.Clusters(namespace).Get(name)
	if err != nil {
		if errors.IsNotFound(err) {
			runtime.HandleError(fmt.Errorf("foo '%s' in work queue no longer exists", key))
			return nil
		}
		return err
	}

	c.Infof("Object: %#v", cluster)

	masterServiceName := fmt.Sprintf("%v-master-service", cluster.Name)
	_, err = c.serviceLister.Services(cluster.Namespace).Get(masterServiceName)
	if errors.IsNotFound(err) {
		_, err = c.kubeclientset.CoreV1().Services(cluster.Namespace).Create(newMasterService(cluster))
	}

	if err != nil {
		return err
	}
<commit_msg>Log creation master discovery service<commit_after>package controller

import (
	"context"
	"fmt"
	"time"

	clientset "github.com/matt-tyler/elasticsearch-operator/pkg/client/clientset/versioned"
	informers "github.com/matt-tyler/elasticsearch-operator/pkg/client/informers/externalversions"
	listers "github.com/matt-tyler/elasticsearch-operator/pkg/client/listers/es/v1"
	"github.com/matt-tyler/elasticsearch-operator/pkg/log"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/util/runtime"

	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	kubeinformers "k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	corelisters "k8s.io/client-go/listers/core/v1"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
)

const maxRetries = 5

type Controller struct {
	log.Logger

	kubeclientset kubernetes.Interface
	esclientset   clientset.Interface

	kubeInformerFactory kubeinformers.SharedInformerFactory
	esInformerFactory   informers.SharedInformerFactory

	clustersSynced cache.InformerSynced
	servicesSynced cache.InformerSynced

	clusterLister listers.ClusterLister
	serviceLister corelisters.ServiceLister

	queue workqueue.RateLimitingInterface
}

func NewController(config *rest.Config) *Controller {
	queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())

	kubeclientset := kubernetes.NewForConfigOrDie(config)
	esclientset := clientset.NewForConfigOrDie(config)

	resyncPeriod := 0 * time.Second

	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeclientset, resyncPeriod)
	esInformerFactory := informers.NewSharedInformerFactory(esclientset, resyncPeriod)

	clusterInformer := esInformerFactory.Es().V1().Clusters()
	serviceInformer := kubeInformerFactory.Core().V1().Services()

	clusterInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			if key, err := cache.MetaNamespaceKeyFunc(obj); err == nil {
				queue.Add(key)
			}
		},
		UpdateFunc: func(oldObj interface{}, newObj interface{}) {
			if key, err := cache.MetaNamespaceKeyFunc(newObj); err == nil {
				queue.Add(key)
			}
		},
		DeleteFunc: func(obj interface{}) {
			if key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj); err == nil {
				queue.Add(key)
			}
		},
	})

	ctx := context.Background()

	go kubeInformerFactory.Start(ctx.Done())
	go esInformerFactory.Start(ctx.Done())

	return &Controller{
		log.NewLogger(),
		kubeclientset,
		esclientset,
		kubeInformerFactory,
		esInformerFactory,
		clusterInformer.Informer().HasSynced,
		serviceInformer.Informer().HasSynced,
		clusterInformer.Lister(),
		serviceInformer.Lister(),
		queue,
	}
}

func (c *Controller) runWorker() {
	for c.processNextItem() {
		// continue looping
	}
}

func (c *Controller) processNextItem() bool {
	obj, shutdown := c.queue.Get()

	if shutdown {
		return false
	}

	err := func(obj interface{}) error {
		defer c.queue.Done(obj)

		var key string
		var ok bool

		if key, ok = obj.(string); !ok {
			c.queue.Forget(obj)
			runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
			return nil
		}

		if err := c.sync(key); err != nil {
			return fmt.Errorf("error syncing '%s': %s", key, err.Error())
		}

		c.queue.Forget(obj)
		return nil
	}(obj)

	if err != nil {
		runtime.HandleError(err)
	}

	return true
}

func (c *Controller) sync(key string) error {
	c.Infof("Processing change to %s", key)

	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		runtime.HandleError(fmt.Errorf("invalid resource key: %s", key))
		return nil
	}

	cluster, err := c.clusterLister.Clusters(namespace).Get(name)
	if err != nil {
		if errors.IsNotFound(err) {
			runtime.HandleError(fmt.Errorf("foo '%s' in work queue no longer exists", key))
			return nil
		}
		return err
	}

	c.Infof("Object: %#v", cluster)

	c.Infof("create master discovery service...")
	masterServiceName := fmt.Sprintf("%v-master-service", cluster.Name)
	_, err = c.serviceLister.Services(cluster.Namespace).Get(masterServiceName)
	if errors.IsNotFound(err) {
		_, err = c.kubeclientset.CoreV1().Services(cluster.Namespace).Create(newMasterService(cluster))
	}

	if err != nil {
		return err
	}

{'fraction_non_alphanumeric': 0.13032699799315312, 'fraction_numerical': 0.004013693778774643, 'mean_word_length': 4.7320703653585925, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4012077', 'n_tokens_mistral': 3156, 'n_tokens_neox': 2947, 'n_words': 524}