// Copyright 2018 PingCAP, 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,
// See the License for the specific language governing permissions and
// limitations under the License.

package tepcluster

import (
	"fmt"
	"time"

	"github.com/golang/glog"
	"github.com/pingcap/tidb-enterprise/installer/tep-operator/pkg/apis/pingcap.com/v1alpha1"
	"github.com/pingcap/tidb-enterprise/installer/tep-operator/pkg/client/clientset/versioned"
	informers "github.com/pingcap/tidb-enterprise/installer/tep-operator/pkg/client/informers/externalversions"
	listers "github.com/pingcap/tidb-enterprise/installer/tep-operator/pkg/client/listers/pingcap.com/v1alpha1"
	"github.com/pingcap/tidb-enterprise/installer/tep-operator/provisioner/ansible"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	eventv1 "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
)

// controllerKind contains the schema.GroupVersionKind for this controller type.
var controllerKind = v1alpha1.SchemeGroupVersion.WithKind("TepCluster")

// Controller controls tepclusters.
type Controller struct {
	// kubernetes client interface
	kubeClient kubernetes.Interface
	// operator client interface
	cli versioned.Interface
	// control returns an interface capable of syncing a tep cluster.
	// Abstracted out for testing.
	control ControlInterface
	// tcLister is able to list/get tepclusters from a shared informer's store
	tcLister listers.TepClusterLister
	// tcListerSynced returns true if the tepcluster shared informer has synced at least once
	tcListerSynced cache.InformerSynced
	// tepclusters that need to be synced.
	queue workqueue.RateLimitingInterface
}

// NewController creates a tepcluster controller.
func NewController(
	kubeCli kubernetes.Interface,
	cli versioned.Interface,
	informerFactory informers.SharedInformerFactory,
) *Controller {
	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartLogging(glog.Infof)
	eventBroadcaster.StartRecordingToSink(&eventv1.EventSinkImpl{
		Interface: eventv1.New(kubeCli.CoreV1().RESTClient()).Events("")})
	recorder := eventBroadcaster.NewRecorder(v1alpha1.Scheme, corev1.EventSource{Component: "tepcluster"})

	tcInformer := informerFactory.Pingcap().V1alpha1().TepClusters()
	statusUpdater := NewRealTepClusterControl(cli, tcInformer.Lister(), recorder)

	tcc := &Controller{
		kubeClient: kubeCli,
		cli:        cli,
		control: NewDefaultTepClusterControl(
			ansible.NewAnsibleProvisoner(),
			statusUpdater,
			recorder,
		),
		queue: workqueue.NewNamedRateLimitingQueue(
			workqueue.DefaultControllerRateLimiter(),
			"tepcluster",
		),
	}

	tcInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: tcc.enqueueTepCluster,
		UpdateFunc: func(old, cur interface{}) {
			tcc.enqueueTepCluster(cur)
		},
		DeleteFunc: tcc.enqueueTepCluster,
	})
	tcc.tcLister = tcInformer.Lister()
	tcc.tcListerSynced = tcInformer.Informer().HasSynced

	return tcc
}

// Run runs the tepcluster controller.
func (tcc *Controller) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer tcc.queue.ShutDown()

	glog.Info("Starting tepcluster controller")
	defer glog.Info("Shutting down tepcluster controller")

	if !cache.WaitForCacheSync(stopCh, tcc.tcListerSynced) {
		return
	}
	wait.Until(tcc.worker, time.Second, stopCh)
}

// worker runs a worker goroutine that invokes processNextWorkItem until the the controller's queue is closed
func (tcc *Controller) worker() {
	for tcc.processNextWorkItem() {
		// revive:disable:empty-block
	}
}

// processNextWorkItem dequeues items, processes them, and marks them done. It enforces that the syncHandler is never
// invoked concurrently with the same key.
func (tcc *Controller) processNextWorkItem() bool {
	key, quit := tcc.queue.Get()
	if quit {
		return false
	}
	defer tcc.queue.Done(key)
	if err := tcc.sync(key.(string)); err != nil {
		utilruntime.HandleError(fmt.Errorf("TepCluster: %v, sync failed %v, requeuing", key.(string), err))
		tcc.queue.AddRateLimited(key)
	} else {
		tcc.queue.Forget(key)
	}
	return true
}

// sync syncs the given tepcluster.
func (tcc *Controller) sync(key string) error {
	startTime := time.Now()
	defer func() {
		glog.V(4).Infof("Finished syncing TepCluster %q (%v)", key, time.Since(startTime))
	}()

	_, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		return err
	}
	tc, err := tcc.tcLister.Get(name)
	if errors.IsNotFound(err) {
		glog.Infof("TepCluster has been deleted %v", key)
		return nil
	}
	if err != nil {
		return err
	}

	return tcc.syncTepCluster(tc.DeepCopy())
}

func (tcc *Controller) syncTepCluster(tc *v1alpha1.TepCluster) error {
	return tcc.control.UpdateTepCluster(tc)
}

// enqueueTepCluster enqueues the given tepcluster in the work queue.
func (tcc *Controller) enqueueTepCluster(obj interface{}) {
	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("Cound't get key for object %+v: %v", obj, err))
		return
	}
	tcc.queue.Add(key)
}