// 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 main

import (
	"flag"
	"fmt"
	"os"
	"time"

	_ "net/http/pprof"

	"github.com/ngaut/log"
	"github.com/pingcap/tidb-operator/pkg/backup"
	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	"github.com/pingcap/tidb-operator/pkg/crd"
	"github.com/pingcap/tidb-operator/pkg/review"
	syncc "github.com/pingcap/tidb-operator/pkg/sync/controller"
	tcc "github.com/pingcap/tidb-operator/pkg/tidbcluster/controller"
	"github.com/pingcap/tidb-operator/pkg/tidbset/controller/pd"
	"github.com/pingcap/tidb-operator/pkg/tidbset/controller/tidb"
	"github.com/pingcap/tidb-operator/pkg/tidbset/controller/tikv"
	"github.com/pingcap/tidb-operator/pkg/util"
	"github.com/pingcap/tidb-operator/pkg/util/election"
	"github.com/pingcap/tidb-operator/pkg/util/election/resourcelock"
	"github.com/pingcap/tidb-operator/version"
	"k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/record"
)

var (
	logLevel       string
	printVersion   bool
	kubeconfig     string
	workers        int
	rbacEnabled    bool
	testMode       bool
	resyncDuration time.Duration

	leaseDuration = 15 * time.Second
	renewDuration = 5 * time.Second
	retryPeriod   = 3 * time.Second
)

func init() {
	flag.StringVar(&logLevel, "L", "info", "log level: debug, info, warn, error, fatal")
	flag.BoolVar(&printVersion, "V", false, "Show version and quit")
	flag.BoolVar(&printVersion, "version", false, "Show version and quit")
	flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to kubeconfig file, omit this if run in cluster")
	flag.StringVar(&tcc.ExternalTrafficPolicy, "external-traffic-policy", string(v1.ServiceExternalTrafficPolicyTypeCluster), fmt.Sprintf("ExternalTrafficPolicy: %s, %s", v1.ServiceExternalTrafficPolicyTypeCluster, v1.ServiceExternalTrafficPolicyTypeLocal))
	flag.IntVar(&workers, "workers", 5, "The number of workers that are allowed to sync concurrently. Larger number = more responsive management, but more CPU (and network) load")
	flag.BoolVar(&rbacEnabled, "rbac-enabled", false, "Whether k8s cluster RBAC is enabled")
	flag.BoolVar(&testMode, "test-mode", false, "Whether this is a test mode")
	flag.DurationVar(&resyncDuration, "resync-duration", 30*time.Second, "resync duration")
	flag.BoolVar(&review.ReviewEnabled, "review-enabled", false, "Whether review enabled")

	flag.Parse()
	log.SetLevelByString(logLevel)
	log.SetHighlighting(false)
	log.SetFlags(log.Ldate | log.Lmicroseconds | log.Lshortfile)
}

func main() {
	if printVersion {
		version.PrintVersionInfo()
		os.Exit(0)
	}
	version.LogVersionInfo()

	hostName, err := os.Hostname()
	if err != nil {
		log.Fatalf("failed to get hostname: %v", err)
	}

	if testMode {
		log.Info("Entering test mode")
	}

	if tcc.ExternalTrafficPolicy != string(v1.ServiceExternalTrafficPolicyTypeCluster) && tcc.ExternalTrafficPolicy != string(v1.ServiceExternalTrafficPolicyTypeLocal) {
		log.Fatalf("--external-traffic-policy=%s must be: %s or %s", tcc.ExternalTrafficPolicy, v1.ServiceExternalTrafficPolicyTypeCluster, v1.ServiceExternalTrafficPolicyTypeLocal)
	}

	tcc.RBACEnabled = rbacEnabled

	ns := os.Getenv("MY_POD_NAMESPACE")
	if ns == "" {
		log.Fatal("MY_POD_NAMESPACE environment variable not set")
	}

	err = crd.EnsureCRDExists(kubeconfig)
	if err != nil {
		log.Fatalf("failed to ensure CRD exists: %v", err)
	}

	var cfg *rest.Config
	if kubeconfig == "" {
		cfg, err = rest.InClusterConfig()
	} else {
		cfg, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
	}
	if err != nil {
		log.Fatalf("failed to build rest.Config: %v", err)
	}
	cfg.QPS = util.ClientCfgQPS
	cfg.Burst = util.ClientCfgBurst

	cli, err := versioned.NewForConfig(cfg)
	if err != nil {
		log.Fatalf("failed to create Clientset: %v", err)
	}

	kubeCli, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		log.Fatalf("failed to get kubernetes Clientset: %v", err)
	}

	rl := resourcelock.EndpointsLock{
		EndpointsMeta: metav1.ObjectMeta{
			Namespace: ns,
			Name:      "tidb-operator",
		},
		Client: kubeCli,
		LockConfig: resourcelock.ResourceLockConfig{
			Identity:      hostName,
			EventRecorder: &record.FakeRecorder{},
		},
	}

	duration := 5 * time.Second

	// leader election for multiple operators
	go wait.Forever(func() {
		election.RunOrDie(election.LeaderElectionConfig{
			Lock:          &rl,
			LeaseDuration: leaseDuration,
			RenewDeadline: renewDuration,
			RetryPeriod:   retryPeriod,
			Callbacks: election.LeaderCallbacks{
				OnStartedLeading: func(stopCh <-chan struct{}) {
					go wait.Forever(func() { tcc.NewTiDBClusterController(cli, kubeCli, resyncDuration).Run(workers, stopCh) }, duration)
					go wait.Forever(func() { pd.NewController(cli, kubeCli, resyncDuration).Run(workers, stopCh) }, duration)
					go wait.Forever(func() { tikv.NewController(cli, kubeCli, resyncDuration).Run(workers, stopCh) }, duration)
					go wait.Forever(func() { syncc.NewSyncController(cli, kubeCli, testMode).Run(workers, stopCh) }, duration)
					go wait.Forever(func() { tidb.NewController(cli, kubeCli, resyncDuration).Run(workers, stopCh) }, duration)
					wait.Forever(func() { backup.NewController(cli, kubeCli).Run(workers, stopCh) }, duration)
				},
				OnStoppedLeading: func() {
					log.Fatalf("leader election lost")
				},
			},
		})
	}, duration)

	err = util.StartPprofServer()
	if err != nil {
		log.Fatalf("Error when starting pprof server: %v", err)
	}
}