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

package controller

import (
	"fmt"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	"github.com/pingcap/tidb-operator/pkg/util"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
)

// ExternalTrafficPolicy denotes if this Service desires to route external traffic to node-local or cluster-wide endpoints.
var ExternalTrafficPolicy string

func (tcc *TiDBClusterController) syncServiceFor(cluster *v1.TidbCluster) error {
	if err := tcc.syncPDClientServiceFor(cluster); err != nil {
		return errors.Trace(err)
	}
	if err := tcc.syncPDPeerServiceFor(cluster); err != nil {
		return errors.Trace(err)
	}
	if err := tcc.syncTiKVPeerServiceFor(cluster); err != nil {
		return errors.Trace(err)
	}

	return tcc.syncTiDBServiceFor(cluster)
}

func (tcc *TiDBClusterController) syncPDClientServiceFor(cluster *v1.TidbCluster) error {
	ns := cluster.GetNamespace()
	svcName := pdSvcName(cluster.GetName())
	svc, err := tcc.kubeCli.CoreV1().Services(ns).Get(svcName, metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			return tcc.createPDClientServiceFor(cluster)
		}

		return errors.Trace(err)
	}

	if svc.Annotations == nil {
		svc.Annotations = make(map[string]string)
	}

	newAnno := util.AnnPaused(nil, cluster)
	svcType := getServiceType(cluster.Spec.Services, "pd")
	if svc.Annotations[label.AnnPaused] == newAnno[label.AnnPaused] &&
		svc.Spec.Type == svcType {
		return nil
	}

	svc.Annotations[label.AnnPaused] = newAnno[label.AnnPaused]
	svc.Spec.Type = svcType
	_, err = tcc.kubeCli.CoreV1().Services(ns).Update(svc)
	if err != nil {
		log.Errorf("Service[%s/%s] annotation[%s] updated failed: %v", ns, svcName, label.AnnPaused, err)
		return errors.Trace(err)
	}

	log.Infof("Service[%s/%s] annotation[%s] updated successfully", ns, svcName, label.AnnPaused)

	return nil
}

func (tcc *TiDBClusterController) syncPDPeerServiceFor(cluster *v1.TidbCluster) error {
	ns := cluster.GetNamespace()
	svcName := pdPeerSvcName(cluster.GetName())
	svc, err := tcc.kubeCli.CoreV1().Services(ns).Get(svcName, metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			return tcc.createPDPeerServiceFor(cluster)
		}
		return errors.Trace(err)
	}
	newAnno := util.AnnPaused(nil, cluster)
	if svc.Annotations == nil {
		svc.Annotations = make(map[string]string)
	}
	if svc.Annotations[label.AnnPaused] == newAnno[label.AnnPaused] {
		return nil
	}
	svc.Annotations[label.AnnPaused] = newAnno[label.AnnPaused]
	_, err = tcc.kubeCli.CoreV1().Services(ns).Update(svc)
	if err != nil {
		log.Errorf("Service[%s/%s] annotation[%s] updated failed: %v", ns, svcName, label.AnnPaused, err)
		return errors.Trace(err)
	}
	log.Infof("Service[%s/%s] annotation[%s] updated successfully", ns, svcName, label.AnnPaused)
	return nil
}

func (tcc *TiDBClusterController) syncTiKVPeerServiceFor(cluster *v1.TidbCluster) error {
	ns := cluster.GetNamespace()
	svcName := tikvPeerSvcName(cluster.GetName())
	svc, err := tcc.kubeCli.CoreV1().Services(ns).Get(svcName, metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			return tcc.createTiKVPeerServiceFor(cluster)
		}
		return errors.Trace(err)
	}
	newAnno := util.AnnPaused(nil, cluster)
	if svc.Annotations == nil {
		svc.Annotations = make(map[string]string)
	}
	if svc.Annotations[label.AnnPaused] == newAnno[label.AnnPaused] {
		return nil
	}
	svc.Annotations[label.AnnPaused] = newAnno[label.AnnPaused]
	_, err = tcc.kubeCli.CoreV1().Services(ns).Update(svc)
	if err != nil {
		log.Errorf("Service[%s/%s] annotation[%s] updated failed: %v", ns, svcName, label.AnnPaused, err)
		return errors.Trace(err)
	}
	log.Infof("Service[%s/%s] annotation[%s] updated successfully", ns, svcName, label.AnnPaused)
	return nil
}

func (tcc *TiDBClusterController) syncTiDBServiceFor(cluster *v1.TidbCluster) error {
	ns := cluster.GetNamespace()
	name := tidbSvcName(cluster.GetName())

	svc, err := tcc.kubeCli.CoreV1().Services(ns).Get(name, metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			return tcc.createTiDBServiceFor(cluster)
		}

		return errors.Trace(err)
	}

	if svc.Annotations == nil {
		svc.Annotations = make(map[string]string)
	}

	newAnno := util.AnnPaused(nil, cluster)
	var svcType corev1.ServiceType
	if cluster.Spec.Services != nil { // get service type from spec.services if exist
		svcType = getServiceType(cluster.Spec.Services, "tidb")
	} else { // for backward compatible
		switch cluster.Spec.Service {
		case "LoadBalancer":
			svcType = corev1.ServiceTypeLoadBalancer
		case "NodePort":
			svcType = corev1.ServiceTypeNodePort
		default:
			svcType = corev1.ServiceTypeClusterIP
		}
	}

	if svc.Annotations[label.AnnPaused] == newAnno[label.AnnPaused] &&
		svc.Spec.Type == svcType &&
		svc.Spec.ExternalTrafficPolicy == corev1.ServiceExternalTrafficPolicyType(ExternalTrafficPolicy) {
		return nil
	}

	svc.Annotations[label.AnnPaused] = newAnno[label.AnnPaused]
	svc.Spec.Type = svcType
	if svc.Spec.Type == corev1.ServiceTypeNodePort || svc.Spec.Type == corev1.ServiceTypeLoadBalancer {
		svc.Spec.ExternalTrafficPolicy = corev1.ServiceExternalTrafficPolicyType(ExternalTrafficPolicy)
	}

	_, err = tcc.kubeCli.CoreV1().Services(ns).Update(svc)
	if err != nil {
		log.Errorf("Service[%s/%s] updated failed: %v", ns, name, err)
		return errors.Trace(err)
	}

	log.Infof("Service[%s/%s] updated successfully", ns, name)
	return nil
}

func (tcc *TiDBClusterController) createPDPeerServiceFor(cluster *v1.TidbCluster) error {
	clusterName := cluster.GetName()
	ns := cluster.GetNamespace()
	svcName := pdPeerSvcName(clusterName)
	pdLabel := label.New().Cluster(clusterName).PD().Labels()

	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:            svcName,
			Namespace:       ns,
			Labels:          pdLabel,
			OwnerReferences: []metav1.OwnerReference{getOwnerRef(cluster)},
			Annotations:     util.AnnPaused(nil, cluster),
		},
		Spec: corev1.ServiceSpec{
			Ports: []corev1.ServicePort{
				{
					Name:       "client",
					Port:       2379,
					TargetPort: intstr.FromInt(2379),
					Protocol:   corev1.ProtocolTCP,
				},
				{
					Name:       "peer",
					Port:       2380,
					TargetPort: intstr.FromInt(2380),
					Protocol:   corev1.ProtocolTCP,
				},
			},
			ClusterIP: "None",
			Selector:  pdLabel,
		},
	}
	_, err := tcc.kubeCli.CoreV1().Services(ns).Create(svc)
	if err != nil {
		log.Errorf("Service[%s/%s] created failed: %v", ns, svcName, err)
		return errors.Trace(err)
	}
	log.Infof("Service[%s/%s] created successfully", ns, svcName)
	return nil
}

func (tcc *TiDBClusterController) createPDClientServiceFor(cluster *v1.TidbCluster) error {
	clusterName := cluster.GetName()
	ns := cluster.GetNamespace()
	name := pdSvcName(clusterName)
	pdLabel := label.New().Cluster(clusterName).PD().Labels()

	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:            name,
			Namespace:       ns,
			Labels:          pdLabel,
			OwnerReferences: []metav1.OwnerReference{getOwnerRef(cluster)},
			Annotations:     util.AnnPaused(nil, cluster),
		},
		Spec: corev1.ServiceSpec{
			Type: getServiceType(cluster.Spec.Services, "pd"),
			Ports: []corev1.ServicePort{
				{
					Name:       "client",
					Port:       2379,
					TargetPort: intstr.FromInt(2379),
					Protocol:   corev1.ProtocolTCP,
				},
			},
			Selector: pdLabel,
		},
	}

	_, err := tcc.kubeCli.CoreV1().Services(ns).Create(svc)
	if err != nil {
		log.Errorf("Service[%s/%s] created failed: %v", ns, name, err)
		return errors.Trace(err)
	}
	log.Infof("Service[%s/%s] created successfully", ns, name)

	return nil
}

func (tcc *TiDBClusterController) createTiKVPeerServiceFor(cluster *v1.TidbCluster) error {
	clusterName := cluster.GetName()
	ns := cluster.GetNamespace()
	svcName := tikvPeerSvcName(clusterName)
	tikvLabel := label.New().Cluster(clusterName).TiKV().Labels()

	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:            svcName,
			Namespace:       ns,
			Labels:          tikvLabel,
			OwnerReferences: []metav1.OwnerReference{getOwnerRef(cluster)},
			Annotations:     util.AnnPaused(nil, cluster),
		},
		Spec: corev1.ServiceSpec{
			Ports: []corev1.ServicePort{
				{
					Name:       "peer",
					Port:       20160,
					TargetPort: intstr.FromInt(20160),
					Protocol:   corev1.ProtocolTCP,
				},
			},
			ClusterIP: "None",
			Selector:  tikvLabel,
		},
	}
	_, err := tcc.kubeCli.CoreV1().Services(ns).Create(svc)
	if err != nil {
		log.Errorf("Service[%s/%s] created failed: %v", ns, svcName, err)
		return errors.Trace(err)
	}
	log.Infof("Service[%s/%s] created successfully", ns, svcName)
	return nil
}

func (tcc *TiDBClusterController) createTiDBServiceFor(cluster *v1.TidbCluster) error {
	clusterName := cluster.GetName()
	ns := cluster.GetNamespace()
	name := tidbSvcName(clusterName)
	tidbLabel := label.New().Cluster(clusterName).TiDB().Labels()

	var svcType corev1.ServiceType
	if len(cluster.Spec.Services) == 0 { // this is for backward compatible
		switch cluster.Spec.Service {
		case "NodePort":
			svcType = corev1.ServiceTypeNodePort
		case "LoadBalancer":
			svcType = corev1.ServiceTypeLoadBalancer
		default:
			svcType = corev1.ServiceTypeClusterIP
		}
	} else {
		svcType = getServiceType(cluster.Spec.Services, "tidb")
	}

	tidbSvc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:            name,
			Namespace:       ns,
			Labels:          tidbLabel,
			OwnerReferences: []metav1.OwnerReference{getOwnerRef(cluster)},
			Annotations:     util.AnnPaused(nil, cluster),
		},
		Spec: corev1.ServiceSpec{
			Type: svcType,
			Ports: []corev1.ServicePort{
				{
					Name:       "mysql-client",
					Port:       4000,
					TargetPort: intstr.FromInt(4000),
					Protocol:   corev1.ProtocolTCP,
				},
				{
					Name:       "status",
					Port:       10080,
					TargetPort: intstr.FromInt(10080),
					Protocol:   corev1.ProtocolTCP,
				},
			},
			Selector: tidbLabel,
		},
	}

	if tidbSvc.Spec.Type == corev1.ServiceTypeNodePort || tidbSvc.Spec.Type == corev1.ServiceTypeLoadBalancer {
		tidbSvc.Spec.ExternalTrafficPolicy = corev1.ServiceExternalTrafficPolicyType(ExternalTrafficPolicy)
	}

	_, err := tcc.kubeCli.CoreV1().Services(ns).Create(tidbSvc)
	if err != nil {
		log.Errorf("Service[%s/%s] created failed: %v", ns, name, err)
		return errors.Trace(err)
	}

	log.Infof("Service[%s/%s] created successfully", ns, name)
	return nil
}

func pdSvcName(clusterName string) string {
	return fmt.Sprintf("%s-pd", clusterName)
}

func pdPeerSvcName(clusterName string) string {
	return fmt.Sprintf("%s-pd-peer", clusterName)
}

func tikvPeerSvcName(clusterName string) string {
	return fmt.Sprintf("%s-tikv-peer", clusterName)
}

func tidbSvcName(clusterName string) string {
	return fmt.Sprintf("%s-tidb", clusterName)
}