// 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"
	_ "net/http/pprof"
	"os"
	"time"

	"github.com/ngaut/log"
	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	"github.com/pingcap/tidb-operator/pkg/review"
	"github.com/pingcap/tidb-operator/pkg/util"
	"github.com/pingcap/tidb-operator/pkg/volumeprovisioner"
	"github.com/pingcap/tidb-operator/pkg/volumeprovisioner/externalstorage/controller"
	"github.com/pingcap/tidb-operator/version"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
)

var (
	logLevel     string
	printVersion bool
	kubeconfig   string
	pvDir        string
	kubeVersion  string
	workers      int

	duration = 5 * 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(&pvDir, "pvdir", "/tidb", "The directory to create PV-backing directories in")
	flag.StringVar(&kubeVersion, "kubeversion", "v1.7.2", "The kubernetes version")
	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(&review.ReviewEnabled, "review-enabled", false, "Whether review enabled")
	flag.Parse()

	log.SetLevelByString(logLevel)
	log.SetHighlighting(false)
	err := flag.Set("logtostderr", "true")
	if err != nil {
		panic(err)
	}
	log.SetFlags(log.Ldate | log.Lmicroseconds | log.Lshortfile)
}

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

	nodeName := os.Getenv("MY_NODE_NAME")
	if nodeName == "" {
		log.Fatalf("MY_NODE_NAME environment variable not set\n")
	}

	var cfg *rest.Config
	var err error
	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)
	}

	// Create the provisioner: it implements the Provisioner interface expected by
	// the controller
	volumeProvisioner := volumeprovisioner.NewVolumeProvisioner(pvDir, nodeName, cli, kubeCli)

	// Start the provision controller which will dynamically provision hostPath PVs
	pc := controller.NewProvisionController(
		kubeCli,
		volumeprovisioner.ProvisionerName,
		volumeProvisioner,
		kubeVersion,
	)
	go wait.Forever(func() {
		pc.Run(wait.NeverStop)
	}, duration)

	vmc := volumeprovisioner.NewController(kubeCli, cli, nodeName)
	go wait.Forever(func() {
		vmc.Run(workers, wait.NeverStop)
	}, duration)

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