// Copyright 2017 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 controller

import (
	"time"

	"fmt"

	"github.com/ngaut/log"
	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
)

const (
	gcDuration = 5 // gc duration in minute
)

// SyncController is the controller of garbage collector controller
type SyncController struct {
	cli     versioned.Interface
	kubeCli kubernetes.Interface

	podController cache.Controller
	podStore      cache.Store
	podQueue      workqueue.RateLimitingInterface

	pvcController cache.Controller
	pvcStore      cache.Store
	pvcQueue      workqueue.RateLimitingInterface

	pvController cache.Controller
	pvStore      cache.Store
	pvQueue      workqueue.RateLimitingInterface

	testMode bool
}

// NewSyncController creates a new SyncController
func NewSyncController(cli versioned.Interface, kubeCli kubernetes.Interface, testMode bool) *SyncController {
	syncc := &SyncController{
		cli:      cli,
		kubeCli:  kubeCli,
		podQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "podqueue"),
		pvcQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pvcqueue"),
		pvQueue:  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pvqueue"),
		testMode: testMode,
	}

	opts := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Labels(),
		).String(),
	}

	syncc.podStore, syncc.podController = cache.NewInformer(
		&cache.ListWatch{
			ListFunc: cache.ListFunc(func(_ metav1.ListOptions) (runtime.Object, error) {
				return syncc.kubeCli.CoreV1().Pods(metav1.NamespaceAll).List(opts)
			}),
			WatchFunc: cache.WatchFunc(func(_ metav1.ListOptions) (watch.Interface, error) {
				return syncc.kubeCli.CoreV1().Pods(metav1.NamespaceAll).Watch(opts)
			}),
		},
		&corev1.Pod{},
		30*time.Second,
		cache.ResourceEventHandlerFuncs{
			AddFunc: syncc.enqueuePod,
			UpdateFunc: func(old, cur interface{}) {
				syncc.enqueuePod(cur)
			},
			DeleteFunc: syncc.enqueuePod,
		},
	)

	syncc.pvcStore, syncc.pvcController = cache.NewInformer(
		&cache.ListWatch{
			ListFunc: cache.ListFunc(func(_ metav1.ListOptions) (runtime.Object, error) {
				return syncc.kubeCli.CoreV1().PersistentVolumeClaims(metav1.NamespaceAll).List(opts)
			}),
			WatchFunc: cache.WatchFunc(func(_ metav1.ListOptions) (watch.Interface, error) {
				return syncc.kubeCli.CoreV1().PersistentVolumeClaims(metav1.NamespaceAll).Watch(opts)
			}),
		},
		&corev1.PersistentVolumeClaim{},
		30*time.Second,
		cache.ResourceEventHandlerFuncs{
			AddFunc: syncc.enqueuePVC,
			UpdateFunc: func(old, cur interface{}) {
				syncc.enqueuePVC(cur)
			},
			DeleteFunc: syncc.enqueuePVC,
		},
	)

	syncc.pvStore, syncc.pvController = cache.NewInformer(
		&cache.ListWatch{
			ListFunc: cache.ListFunc(func(_ metav1.ListOptions) (runtime.Object, error) {
				return syncc.kubeCli.CoreV1().PersistentVolumes().List(opts)
			}),
			WatchFunc: cache.WatchFunc(func(_ metav1.ListOptions) (watch.Interface, error) {
				return syncc.kubeCli.CoreV1().PersistentVolumes().Watch(opts)
			}),
		},
		&corev1.PersistentVolume{},
		30*time.Second,
		cache.ResourceEventHandlerFuncs{
			AddFunc: syncc.enqueuePV,
			UpdateFunc: func(old, cur interface{}) {
				syncc.enqueuePV(cur)
			},
			DeleteFunc: syncc.enqueuePV,
		},
	)

	return syncc
}

// Run method runs GC controller
func (syncc *SyncController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	defer syncc.podQueue.ShutDown()
	log.Infof("Starting GC Pod controller")
	go syncc.podController.Run(stopCh)
	for i := 0; i < workers; i++ {
		go wait.Until(syncc.podWorker, time.Second, stopCh)
	}

	defer syncc.pvcQueue.ShutDown()
	log.Infof("Starting GC PVC controller")
	go syncc.pvcController.Run(stopCh)
	for i := 0; i < workers; i++ {
		go wait.Until(syncc.pvcWorker, time.Second, stopCh)
	}

	defer syncc.pvQueue.ShutDown()
	log.Infof("Starting GC PV controller")
	go syncc.pvController.Run(stopCh)
	for i := 0; i < workers; i++ {
		go wait.Until(syncc.pvWorker, time.Second, stopCh)
	}

	<-stopCh
	log.Infof("Shutting down SyncController")
}

func (syncc *SyncController) podWorker() {
	for {
		func() {
			key, quit := syncc.podQueue.Get()
			if quit {
				return
			}
			defer syncc.podQueue.Done(key)

			if err := syncc.syncPod(key.(string)); err != nil {
				utilruntime.HandleError(fmt.Errorf("Error syncing Pod %v, requeuing: %v", key.(string), err))
				syncc.podQueue.AddRateLimited(key)
			} else {
				syncc.podQueue.Forget(key)
			}
		}()
	}
}

func (syncc *SyncController) enqueuePod(obj interface{}) {
	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	if err != nil {
		log.Errorf("can't get key for obj: %v, error: %v", obj, err)
		return
	}

	syncc.podQueue.Add(key)
}

func (syncc *SyncController) pvcWorker() {
	for {
		func() {
			key, quit := syncc.pvcQueue.Get()
			if quit {
				return
			}
			defer syncc.pvcQueue.Done(key)

			if err := syncc.syncPVC(key.(string)); err != nil {
				utilruntime.HandleError(fmt.Errorf("Error syncing PVC %v, requeuing: %v", key.(string), err))
				syncc.pvcQueue.AddRateLimited(key)
			} else {
				syncc.pvcQueue.Forget(key)
			}
		}()
	}
}

func (syncc *SyncController) enqueuePVC(obj interface{}) {
	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	if err != nil {
		log.Errorf("can't get key for obj: %v, error: %v", obj, err)
		return
	}

	syncc.pvcQueue.Add(key)
}

func (syncc *SyncController) pvWorker() {
	for {
		func() {
			key, quit := syncc.pvQueue.Get()
			if quit {
				return
			}
			defer syncc.pvQueue.Done(key)

			if err := syncc.syncPV(key.(string)); err != nil {
				utilruntime.HandleError(fmt.Errorf("Error syncing PV %v, requeuing: %v", key.(string), err))
				syncc.pvQueue.AddRateLimited(key)
			} else {
				syncc.pvQueue.Forget(key)
			}
		}()
	}
}

func (syncc *SyncController) enqueuePV(obj interface{}) {
	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	if err != nil {
		log.Errorf("can't get key for obj: %v, error: %v", obj, err)
		return
	}

	syncc.pvQueue.Add(key)
}