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

import (
	"bufio"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/exec"
	"os/user"
	"path"
	"strconv"
	"strings"
	"sync"
	"unicode"

	"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/utils"
	"github.com/pingcap/tidb-enterprise/installer/tep-operator/provisioner"
)

var WorkDir string

// Provisioner is used to complete specific tasks through ansible
type Provisioner struct {
	config  *v1alpha1.AnsibleConfig
	workDir string
	runtime *exec.Cmd
	status  provisioner.ProvisionerStatus
}

// NewAnsibleProvisoner returns a new instance that implement Provisioner Interface
func NewAnsibleProvisoner() provisioner.Provisioner {
	return &Provisioner{
		workDir: WorkDir,
	}
}

func (p *Provisioner) Prepare(tc *v1alpha1.TepCluster) error {
	glog.Info("Prepare for deploy tep cluster")
	err := p.setAnsibleCfg(tc)
	if err != nil {
		return err
	}
	if p.needGenerateInventoryFile(tc) {
		err = p.generateInventoryFile(tc)
		if err != nil {
			return err
		}
	}
	if tc.Spec.Launch != v1alpha1.TepClusterPrepare {
		return nil
	}
	p.status.Phase = v1alpha1.DeployPreparing
	//TODO: collect hosts's disk info
	return nil
}

func (p *Provisioner) Provision(tc *v1alpha1.TepCluster) error {
	if tc.Spec.Launch != v1alpha1.TepClusterInstall || tc.Status.Phase != v1alpha1.DeployPreparing {
		return nil
	}
	glog.Info("Provisioning TEP Cluster with Ansible...")
	//TODO: need to consider the tep cluster's scale out and scale in situation later
	p.status.Phase = v1alpha1.DeployProgress

	go func() {
		glog.Infof("start to executing ansible playbook %s", p.config.PlaybookFile)
		err := p.executeAnsible()
		if err != nil {
			p.status.Phase = v1alpha1.DeployFailed
			glog.Error(err)
			return
		}
		p.status.Phase = v1alpha1.DeploySucceeded
	}()
	return nil
}

func (p *Provisioner) Cancel() error {
	if p.runtime == nil {
		return nil
	}
	err := p.runtime.Process.Kill()
	if err != nil {
		return fmt.Errorf("stop tep cluster failed, phase: %s, err: %s", p.status.Phase, err.Error())
	}
	p.status.Phase = v1alpha1.DeployFailed
	return nil
}

func (p *Provisioner) GetProvisionerStatus() provisioner.ProvisionerStatus {
	return p.status
}

// needGenerateInventoryFile indicates whether the inventory file needs to be regenerated
func (p *Provisioner) needGenerateInventoryFile(tc *v1alpha1.TepCluster) bool {
	if (tc.Spec.Launch == v1alpha1.TepClusterPrepare && len(p.config.InventoryFile) == 0) ||
		(tc.Spec.Launch == v1alpha1.TepClusterInstall && tc.Status.Phase == v1alpha1.DeployPreparing) {
		return true
	}
	return false
}

func (p *Provisioner) generateInventoryFile(tc *v1alpha1.TepCluster) error {
	tmpInventoryFile, err := ioutil.TempFile(p.config.InventoryDirectory, "provisioner-ansible-")
	if err != nil {
		return fmt.Errorf("error preparing inventory file: %s", err.Error())
	}
	defer tmpInventoryFile.Close()
	w := bufio.NewWriter(tmpInventoryFile)
	for group, hosts := range tc.Spec.GroupMapHosts {
		fmt.Fprintf(w, "[%s]\n", group)
		for _, host := range hosts {
			item := []string{
				fmt.Sprintf("%s ansible_ssh_user=%s ansible_ssh_port=%s",
					host.HostName, p.config.User, p.config.Port),
			}
			if len(host.K8sLabels) != 0 {
				item = append(item, fmt.Sprintf("labels=%s", host.K8sLabels))
			}
			if len(host.HostVars) != 0 {
				item = append(item, host.HostVars.String())
			}
			w.WriteString(strings.Join(item, " "))
			w.WriteString("\n")
		}
		fmt.Fprint(w, "\n")
	}
	// Aggregate the kube-master and kube-node groups into the k8s-cluster group
	fmt.Fprint(w, "[k8s-cluster:children]\n")
	fmt.Fprint(w, "kube-master\n")
	fmt.Fprint(w, "kube-node\n")
	if err := w.Flush(); err != nil {
		return fmt.Errorf("error preparing inventory file: %s", err)
	}
	p.config.InventoryFile = tmpInventoryFile.Name()
	return nil
}

func (p *Provisioner) setAnsibleCfg(tc *v1alpha1.TepCluster) error {
	if tc.Spec.Ansible == nil {
		p.config = &v1alpha1.AnsibleConfig{
			Command: "ansible-playbook",
		}
	} else {
		p.config = tc.Spec.Ansible.DeepCopy()
	}
	// Default
	if p.config.Command == "" {
		p.config.Command = "ansible-playbook"
	}

	if p.config.InventoryDirectory == "" {
		p.config.InventoryDirectory = path.Join(p.workDir, "inventory")
	}
	err := utils.ValidateDirectoryConfig(p.config.InventoryDirectory)
	if err != nil {
		return err
	}

	playbookFile, err := p.getPlaybookFile(tc.Spec.Launch)
	if err != nil {
		return err
	}
	p.config.PlaybookFile = playbookFile

	if p.config.User == "" {
		usr, err := user.Current()
		if err != nil {
			return fmt.Errorf("user: could not determine current user from environment, error: %s", err.Error())
		}
		p.config.User = usr.Username
	}
	if len(p.config.Port) > 0 {
		if _, err := strconv.ParseUint(p.config.Port, 10, 16); err != nil {
			return fmt.Errorf(" remote ssh port: %s must be a valid port", p.config.Port)
		}
	} else {
		p.config.Port = "22"
	}
	return nil
}

func (p *Provisioner) getPlaybookFile(launchType v1alpha1.TepClusterLaunchType) (string, error) {
	var playbookFile string
	var require bool
	switch launchType {
	case v1alpha1.TepClusterInstall:
		playbookFile = path.Join(p.workDir, "playbooks", "deploy_cluster.yml")
		require = true
	case v1alpha1.TepClusterPrepare:
		playbookFile = path.Join(p.workDir, "playbooks", "prepare.yml")
	}
	return playbookFile, utils.ValidateFileConfig(playbookFile, "playbook_file", require)
}

func (p *Provisioner) executeAnsible() error {
	inventory := p.config.InventoryFile
	var envvars []string

	args := []string{"-i", inventory, p.config.PlaybookFile}
	args = append(args, p.config.ExtraArguments...)
	if len(p.config.AnsibleEnvVars) > 0 {
		envvars = append(envvars, p.config.AnsibleEnvVars...)
	}

	if len(p.config.ExtraVars) > 0 {
		args = append(args, "--extra-vars", strings.Join(p.config.ExtraVars, " "))
	}

	p.runtime = exec.Command(p.config.Command, args...)
	p.runtime.Dir = p.workDir

	p.runtime.Env = os.Environ()
	if len(envvars) > 0 {
		p.runtime.Env = append(p.runtime.Env, envvars...)
	}

	stdout, err := p.runtime.StdoutPipe()
	if err != nil {
		return err
	}
	stderr, err := p.runtime.StderrPipe()
	if err != nil {
		return err
	}

	wg := &sync.WaitGroup{}
	repeat := func(r io.ReadCloser) {
		reader := bufio.NewReader(r)
		for {
			line, err := reader.ReadString('\n')
			if line != "" {
				line = strings.TrimRightFunc(line, unicode.IsSpace)
				fmt.Println(line)
			}
			if err != nil {
				if err == io.EOF {
					break
				}
				glog.Error(err.Error())
				break
			}
		}
		wg.Done()
	}
	wg.Add(2)
	go repeat(stdout)
	go repeat(stderr)

	glog.Infof("Executing Ansible: %s", strings.Join(p.runtime.Args, " "))
	if err := p.runtime.Start(); err != nil {
		return fmt.Errorf("start ansible playbook %s failed, error: %s", p.config.PlaybookFile, err.Error())
	}
	wg.Wait()
	err = p.runtime.Wait()
	if err != nil {
		return fmt.Errorf("executing ansible playbook %s failed: %s", p.config.PlaybookFile, err.Error())
	}

	return nil
}