package k8s

import (
	"context"
	"crypto"
	"crypto/ecdsa"
	"crypto/rand"
	"crypto/rsa"
	"crypto/x509"
	"crypto/x509/pkix"
	"encoding/base64"
	"encoding/json"
	"encoding/pem"
	"fmt"
	"io/ioutil"
	"net/url"
	"strings"
	"time"

	"github.com/hashicorp/go-multierror"
	pkgErrors "github.com/pkg/errors"
	batchv1 "k8s.io/api/batch/v1"
	v1 "k8s.io/api/certificates/v1"
	"k8s.io/api/certificates/v1beta1"
	corev1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	kerrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/util/retry"
	"k8s.io/utils/pointer"

	"github.com/pingcap/test-infra/common/errors"
	"github.com/pingcap/test-infra/common/model"
	mResource "github.com/pingcap/test-infra/common/model/resource"
	"github.com/pingcap/test-infra/common/model/rms/request"
	"github.com/pingcap/test-infra/common/model/rms/response"
	"github.com/pingcap/test-infra/sdk/core"
	sdkResouce "github.com/pingcap/test-infra/sdk/resource"
	"github.com/pingcap/test-infra/sdk/resource/impl/k8s"
	rmsConfig "github.com/pingcap/test-infra/services/rms/config"
	"github.com/pingcap/test-infra/services/rms/log"
	"github.com/pingcap/test-infra/services/rms/metrics"
	"github.com/pingcap/test-infra/services/rms/service/provider"
	"github.com/pingcap/test-infra/services/rms/service/provider/k8s/gc"
	"github.com/pingcap/test-infra/services/rms/utils"
)

var dateFormat = "2006/01/02 15:04:05.000 -07:00"

var (
	_ provider.Interface = &DefaultProvider{}
)

type DefaultProvider struct {
	clientConfig  *rest.Config
	apiserverURL  *url.URL
	kubeConfig    string
	ClientSet     kubernetes.Interface
	DynamicClient dynamic.Interface

	TiDBCloud k8s.TiDBCloudKubeClientSet
}

func NewDefaultProvider(kubeConfig string) (*DefaultProvider, error) {
	// setup sdk
	k8s.SetConfig(rmsConfig.DefaultRMSConfiguration.K8s.Config)

	clientConfig, err := k8s.LoadConfig(kubeConfig)
	if err != nil {
		return nil, err
	}
	clientConfig.QPS = rmsConfig.DefaultRMSConfiguration.K8s.RequestQPS // token bucket ratelimit config
	clientConfig.Burst = rmsConfig.DefaultRMSConfiguration.K8s.RequestBurst

	apiserverURL, _ := url.Parse(rmsConfig.DefaultRMSConfiguration.K8s.K8sApiserver)

	clientSet, err := kubernetes.NewForConfig(clientConfig)
	if err != nil {
		return nil, err
	}
	dynamicClient, err := dynamic.NewForConfig(clientConfig)
	if err != nil {
		return nil, err
	}

	var tidbCloudClientset k8s.TiDBCloudKubeClientSet
	tidbCloudConfig := rmsConfig.DefaultRMSConfiguration.TiDBCloudConfiguration
	if len(tidbCloudConfig.BaseEKSAPIServerEndpoint) != 0 { // naive way to decide if tidbCloudConfig empty
		tidbCloudClientset, err = k8s.NewTiDBCloudKubeClientSet(&tidbCloudConfig)
		if err != nil {
			return nil, err
		}
	} else { // empty tidbCloudConfig, use fake implementation as place holder
		tidbCloudClientset = k8s.NewFakeTiDBCloudClient()
	}

	p := &DefaultProvider{
		clientConfig:  clientConfig,
		apiserverURL:  apiserverURL,
		kubeConfig:    kubeConfig,
		ClientSet:     clientSet,
		DynamicClient: dynamicClient,
		TiDBCloud:     tidbCloudClientset,
	}

	nss, err := p.getCurrentTestbeds()
	if err != nil {
		return nil, fmt.Errorf("load existing testbeds fail: %w", err)
	}
	for _, ns := range nss {
		metrics.NewTestbed(ns.Name)
	}
	return p, nil
}

func getAccounting(ctx utils.Context, auth request.TestBedMeta) string {
	user := utils.GetUser(ctx)
	if user == nil {
		return ""
	}
	if !user.IsAdmin() {
		return generateAccountingNameFor(user)
	}
	return fmt.Sprintf("%s-%s", auth.Accounting.Kind, auth.Accounting.Name)
}

func generateAccountingNameFor(user *utils.User) string {
	return fmt.Sprintf("USER-%s", base64.RawURLEncoding.EncodeToString([]byte(user.Name)))
}

func (p *DefaultProvider) CreateTestbed(ctx utils.Context, testbedReq request.CreateTestbedRequest) (*mResource.TestBed, *errors.StatusError) {
	if err := p.checkQuotaLimitOfUser(ctx, utils.GetUser(ctx)); err != nil {
		return nil, err
	}
	// create a namespace for the testbed and linked sub resources
	ns := &corev1.Namespace{
		ObjectMeta: metav1.ObjectMeta{
			Name:         testbedReq.Name,
			GenerateName: testbedReq.GenerateName,
			Labels: map[string]string{
				"app.kubernetes.io/managed-by": "test-infra",
				"accounting":                   getAccounting(ctx, testbedReq.TestbedMeta),
			},
			Annotations: map[string]string{
				// used by admission-control plugin: PodTolerationRestriction (test-infra resrouce-pool)
				"scheduler.alpha.kubernetes.io/defaultTolerations": `[{"operator": "Equal", "effect": "NoSchedule", "key": "dedicated", "value": "test-infra"}]`,
				// used by admission-control plugin: PodNodeSelector (test-infra resrouce-pool)
				"scheduler.alpha.kubernetes.io/node-selector": fmt.Sprintf("resourcepool=%s", testbedReq.ResourcePool),
			},
		},
	}
	p.setTestbedKubeScheduler(ns)

	err := gc.SetLeaseInfoOn(ns, time.Now(), testbedReq.ReclaimPolicy.Lease)
	if err != nil {
		return nil, errors.NewInternalError(err)
	}

	ns, err = p.ClientSet.CoreV1().Namespaces().Create(ctx.Request().Context(), ns, metav1.CreateOptions{})
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("create testbed error: %s", err))
	} else if testbedReq.Name == "" {
		testbedReq.Name = ns.Name
	}
	if err = p.bindingRoleOnServiceAccountDefault(ctx, ns); err != nil {
		return nil, errors.NewInternalError(err)
	}
	err = p.authorizeTestbedToUsers(ctx, testbedReq.TestbedMeta.AuthorizeTo, *ns, rmsConfig.DefaultRMSConfiguration.K8s.AuthorizeRole)
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("authorize testbed error: %s", err))
	}
	testbedClient := k8s.NewTestbedClient(k8s.NewKubeClient(ns.Name, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	// construct the testbed
	spec, err := json.Marshal(testbedReq)
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("create testbed error: %s", err))
	}
	testbed, err := testbedClient.Create(spec)
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("create testbed error: %s", err))
	}
	metrics.NewTestbed(testbed.Handle().Name)
	return &mResource.TestBed{ResourceHandle: testbed.Handle(), Qos: testbedReq.Qos}, nil
}

// setTestbedKubeScheduler sets schedulerName annotation to namespace of testbed
// then mutating webhook will set pod.spec.schedulerName field of all pods under the namespace
func (p *DefaultProvider) setTestbedKubeScheduler(ns *corev1.Namespace) *corev1.Namespace {
	if len(rmsConfig.DefaultRMSConfiguration.K8s.KubeSchedulerName) > 0 {
		// see: https://github.com/pingcap/test-infra/blob/main/services/webhook/pkg/mutation/scheduler_name.go
		ns.Annotations["test-infra.pingcap.net/schedulerName"] = rmsConfig.DefaultRMSConfiguration.K8s.KubeSchedulerName
	}

	return ns
}

func (p *DefaultProvider) bindingRoleOnServiceAccountDefault(ctx utils.Context, ns *corev1.Namespace) error {
	// bind the sa default with the cluster-admin role
	if _, err := p.ClientSet.RbacV1().RoleBindings(ns.Name).Create(ctx.Request().Context(), &rbacv1.RoleBinding{
		ObjectMeta: metav1.ObjectMeta{
			GenerateName: "rb-",
		},
		Subjects: []rbacv1.Subject{
			{
				Kind:      "ServiceAccount",
				Name:      "default",
				Namespace: ns.Name,
			},
		},
		RoleRef: rbacv1.RoleRef{
			Kind:     "ClusterRole",
			Name:     "cluster-admin",
			APIGroup: "rbac.authorization.k8s.io",
		},
	}, metav1.CreateOptions{}); err != nil {
		return pkgErrors.Wrap(err, "create a rolebinding clusterrole=cluster-admin on serviceAccount default error")
	}
	return nil
}

func (p *DefaultProvider) GetShootKubeConfig(ctx utils.Context, req request.GetShootKubeConfigRequest) (*response.GetShootKubeConfigResponse, *errors.StatusError) {
	kubeConfig, err := p.TiDBCloud.GetShootKubeConfig(ctx.Request().Context(), req.ProjectID, req.CloudProvider, req.Region)
	if err != nil {
		return nil, errors.NewInternalError(err)
	}

	return &response.GetShootKubeConfigResponse{KubeConfig: kubeConfig}, nil
}

func (p *DefaultProvider) GetTestbed(ctx utils.Context, testbedName string, withDetails bool) (*response.GetTestbedResponse, *errors.StatusError) {
	if _, err := p.checkNamespace(ctx.Request().Context(), testbedName); err != nil {
		return nil, err
	}
	resp := &response.GetTestbedResponse{
		Name: testbedName,
	}

	testbedSpec, err := p.getSpec(ctx, testbedName)
	if err != nil {
		return nil, err
	}
	for _, item := range testbedSpec.Items {
		respItem, e := p.GetResource(ctx,
			mResource.ResourceHandle{
				Name:    item.Name,
				Testbed: testbedName,
				Type:    item.Type,
			},
			withDetails)
		if e != nil {
			return nil, e
		}
		resp.Items = append(resp.Items, respItem)

	}

	resp.Lease, err = p.GetLease(ctx, testbedName)
	if err != nil {
		ctx.Log().Errorf(err, "get lease error, testbed: %s", testbedName)
		return nil, err
	}

	// block TiDBCluster from being ready if any holder pod isn't ready
	// TODO: use better approach, we might be able to use holder CRD, then code logic won't be so coupled
	holderPodReady, err := p.isHolderPodReady(ctx, testbedName)
	if err != nil {
		return nil, err
	}
	if !holderPodReady {
		for i := range resp.Items {
			if resp.Items[i].Type == mResource.ResourceTypeTiDBCluster {
				resp.Items[i].Ready = &[]bool{false}[0]
			}
		}
	}

	return resp, nil
}

func (p *DefaultProvider) isHolderPodReady(ctx utils.Context, testbedName string) (bool, *errors.StatusError) {
	holderPods, err := p.ClientSet.CoreV1().Pods(testbedName).List(ctx.Request().Context(), metav1.ListOptions{LabelSelector: labels.Set{
		"app.kubernetes.io/component": "test-infra-holder",
	}.AsSelector().String()})
	if err != nil {
		return false, errors.NewInternalError(err)
	}

	isReady := true
	for _, holderPod := range holderPods.Items {
		if holderPod.Status.Phase != corev1.PodRunning {
			isReady = false
			break
		}
	}

	return isReady, nil
}

func (p *DefaultProvider) CheckTestbed(ctx utils.Context, testbedName string) (*response.CheckTestbedResponse, *errors.StatusError) {
	eventList, err := p.ClientSet.CoreV1().Events(testbedName).List(ctx.Request().Context(), metav1.ListOptions{FieldSelector: "type=Warning"})
	if err != nil {
		return nil, errors.NewInternalError(err)
	}

	var errMessages []string
	for _, event := range eventList.Items {
		if isImageFailed(event) {
			errMessages = append(errMessages, fmt.Sprintf("pod %s/%s: image error. %s", event.InvolvedObject.Namespace, event.InvolvedObject.Name, event.Message))
		} else if isPodCrashLoop(event) {
			errMessages = append(errMessages, fmt.Sprintf("pod %s/%s: pod crash loop. %s", event.InvolvedObject.Namespace, event.InvolvedObject.Name, event.Message))
		}
	}
	msg, statusErr := p.checkTiDBClusterPostStartJobCompleted(ctx, testbedName)
	if statusErr != nil {
		return nil, statusErr
	}
	if len(msg) > 0 {
		errMessages = append(errMessages, msg...)
	}
	resp := &response.CheckTestbedResponse{Name: testbedName, Status: model.TestbedHealthy}
	if len(errMessages) > 0 {
		resp.Status = model.TestbedFailed
		resp.Message = strings.Join(errMessages, "\n")
	}

	return resp, nil
}

func (p *DefaultProvider) checkTiDBClusterPostStartJobCompleted(ctx utils.Context, testbedName string) (errMessages []string, statusError *errors.StatusError) {
	setJobs, err := p.ClientSet.BatchV1().Jobs(testbedName).List(ctx.Request().Context(), metav1.ListOptions{
		LabelSelector: labels.Set{
			"app.kubernetes.io/component": "tc-setup",
		}.AsSelector().String(),
	})
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	for _, job := range setJobs.Items {
		for _, condition := range job.Status.Conditions {
			if condition.Type == batchv1.JobFailed && condition.Status == corev1.ConditionTrue {
				errMessages = append(errMessages, fmt.Sprintf("tidb_cluster %s.postStart.sql error, please check your sql statements", job.Labels["app.kubernetes.io/instance"]))
				break
			}
		}
	}
	return errMessages, nil
}

func isImageFailed(event corev1.Event) bool {
	return event.Source.Component == "kubelet" &&
		event.Reason == "Failed" &&
		// values from: https://github.com/kubernetes/kubernetes/blob/master/pkg/kubelet/images/types.go#L27-L43
		(strings.Contains(event.Message, "InvalidImageName") || strings.Contains(event.Message, "ErrImagePull"))
}

func isPodCrashLoop(event corev1.Event) bool {
	return event.Source.Component == "kubelet" &&
		// values from https://github.com/kubernetes/kubernetes/blob/dab6f6a43d5a41ef3e2aa11d4a85c45a268ac6cd/pkg/kubelet/kuberuntime/kuberuntime_manager.go#L954
		event.Reason == "BackOff" && strings.Contains(event.Message, "Back-off restarting failed container") &&
		// treat it as un-recoverable crash if crash backoff event has been seen for 20 minutes and 20 times.
		// (value 20 minutes and 20 times are picked base on our daily-run testbed statistics)
		event.FirstTimestamp.Before(&metav1.Time{Time: time.Now().Add(-20 * time.Minute)}) &&
		event.Count > 20
}

func (p *DefaultProvider) ListTestbed(ctx utils.Context) (*response.ListTestbedResponse, *errors.StatusError) {
	user := utils.GetUser(ctx)
	if user == nil {
		return nil, errors.NewBadRequest("cannot get the user info")
	}

	labelSelector := labels.Set{
		"app.kubernetes.io/managed-by": "test-infra",
	}
	if !user.IsAdmin() {
		labelSelector["accounting"] = generateAccountingNameFor(user)
	}

	// check the testbed quota limit
	nss, err := p.ClientSet.CoreV1().Namespaces().List(ctx.Request().Context(), metav1.ListOptions{
		LabelSelector: labelSelector.AsSelector().String(),
	})
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	resp := &response.ListTestbedResponse{}
	for _, ns := range nss.Items {
		expireTs, lease, err := gc.GetLeaseInfoOn(&ns)
		if err == gc.ErrNoLeaseInfoFound {
			continue
		}
		resp.Items = append(resp.Items, response.GetTestbedResponse{
			Name: ns.Name,
			Lease: &response.LeaseResponse{
				Lease:      lease.String(),
				ExpireDate: expireTs.Format(dateFormat),
			},
		})
	}
	return resp, nil
}

func (p *DefaultProvider) GetRestoredMonitorInfo(ctx utils.Context, restoredTestbed string) (*response.RestoreMonitorResponse, *errors.StatusError) {
	serviceList, err := p.ClientSet.CoreV1().Services(restoredTestbed).List(ctx.Request().Context(), metav1.ListOptions{})
	if err != nil {
		return nil, errors.NewInternalError(err)
	}

	for _, service := range serviceList.Items {
		if strings.Contains(service.Name, "grafana") &&
			service.Spec.Type == corev1.ServiceTypeNodePort {
			return &response.RestoreMonitorResponse{
				TestbedName: restoredTestbed,
				GrafanaURL:  fmt.Sprintf("http://%s:%d", p.apiserverURL.Hostname(), service.Spec.Ports[0].NodePort),
			}, nil
		}
	}

	return nil, errors.NewInternalError(fmt.Errorf("Grafana of restored testbed %s  not found", restoredTestbed))
}

func (p *DefaultProvider) authorizeTestbedToUsers(ctx utils.Context, users []request.User, namespace corev1.Namespace, authRole string) error {
	user := utils.GetUser(ctx)
	if user == nil {
		ctx.Log().Infof("not create role-binding for unauthorized user")
		return errors.NewBadRequest("cannot get the user info")
	}

	if !user.IsAdmin() {
		users = []request.User{{Kind: "User", Name: user.Name}}
	}

	roleBindingList, err := p.ClientSet.RbacV1().RoleBindings(namespace.GetName()).List(ctx.Request().Context(), metav1.ListOptions{})
	if err != nil {
		return errors.NewInternalError(err)
	}

	for _, u := range users {
		if roleBindingExist(roleBindingList.Items, u, authRole) {
			continue
		}

		_, err := p.ClientSet.RbacV1().RoleBindings(namespace.Name).Create(ctx.Request().Context(), &rbacv1.RoleBinding{
			ObjectMeta: metav1.ObjectMeta{
				GenerateName: "rb-",
			},
			Subjects: []rbacv1.Subject{
				{
					Kind:      u.Kind,
					Name:      u.Name,
					Namespace: u.Namespace,
				},
			},
			RoleRef: rbacv1.RoleRef{
				Kind:     "ClusterRole",
				Name:     authRole,
				APIGroup: "rbac.authorization.k8s.io",
			},
		}, metav1.CreateOptions{})
		if err != nil {
			return errors.NewInternalError(err)
		}
	}

	return nil
}

func roleBindingExist(roleBindingList []rbacv1.RoleBinding, user request.User, role string) bool {
	for _, roleBinding := range roleBindingList {
		if roleBinding.RoleRef.Kind == "ClusterRole" &&
			roleBinding.RoleRef.Name == role &&
			roleBinding.RoleRef.APIGroup == "rbac.authorization.k8s.io" &&
			roleBinding.Subjects[0].Kind == user.Kind &&
			roleBinding.Subjects[0].Name == user.Name &&
			roleBinding.Subjects[0].Namespace == user.Namespace {
			return true
		}
	}

	return false
}

func (p *DefaultProvider) AuthTestbed(ctx utils.Context, req request.AuthTestbedRequest) *errors.StatusError {
	ns, err := p.ClientSet.CoreV1().Namespaces().Get(ctx.Request().Context(), req.Name, metav1.GetOptions{})
	if err != nil {
		return errors.NewInternalError(err)
	}

	if err = p.authorizeTestbedToUsers(ctx, req.AuthorizeTo, *ns, rmsConfig.DefaultRMSConfiguration.K8s.AuthorizeRoleForDebugger); err != nil {
		return errors.NewInternalError(err)
	}

	return nil
}

func (p *DefaultProvider) DeleteTestbed(ctx utils.Context, name string) *errors.StatusError {
	// since all resources are hosted on a namespace, deleting the namespace could release all of these resources
	user := utils.GetUser(ctx)
	if user == nil {
		return errors.NewBadRequest("cannot get the user info")
	}
	ns, serr := p.checkNamespace(context.Background(), name)
	if serr != nil {
		return serr
	}
	if ns.Labels == nil || ns.Labels["app.kubernetes.io/managed-by"] != "test-infra" {
		return errors.NewNotFound(fmt.Sprintf("testbed %s", name))
	}
	if !strings.HasPrefix(ns.Labels["accounting"], "ID-") {
		// FIXME@codingpoet: should auth with tcms,
		//  Let plan maintainers able to delete testbeds
		// temporarily bypass the testbeds created by tps
		if !user.IsAdmin() && ns.Labels["accounting"] != generateAccountingNameFor(user) {
			return errors.NewForbidden(fmt.Sprintf("delete the testbed %s", name))
		}
	}
	if err := p.ClientSet.CoreV1().Namespaces().Delete(context.Background(), name, metav1.DeleteOptions{}); err != nil {
		return errors.NewInternalError(err)
	}
	metrics.DelTestbed(name)
	return nil
}

func (p *DefaultProvider) OnCompletion(ctx utils.Context, req request.OnTestbedCompletionRequest) *errors.StatusError {
	testbed := req.Name
	completionType := req.Type

	user := utils.GetUser(ctx)
	if user == nil {
		return errors.NewBadRequest("cannot get the user info")
	}

	if _, err := p.checkNamespace(ctx.Request().Context(), testbed); err != nil {
		return err
	}

	strategy, err := p.GetGcStrategy(ctx, testbed)
	if err != nil {
		return err
	}
	switch completionType {
	case request.OnSuccess:
		actions := strategy.OnSuccess
		if actions == nil {
			actions = strategy.OnCompletion
		}
		if actions == nil {
			return errors.NewInternalError(fmt.Errorf("gcStrategy of %s doesn't exist", testbed))
		}
		return p.onComplete(ctx, completionType, testbed, req.ClinicURLs, actions)
	case request.OnFailure:
		actions := strategy.OnFailure
		if actions == nil {
			actions = strategy.OnCompletion
		}
		if actions == nil {
			return errors.NewInternalError(fmt.Errorf("gcStrategy of %s doesn't exist", testbed))
		}
		return p.onComplete(ctx, completionType, testbed, req.ClinicURLs, actions)
	default:
		return errors.NewInternalError(fmt.Errorf("unknown action type: %s", completionType))
	}
}

//nolint:gocyclo,gocognit // TODO: 1. simplify this method 2. use string builder to reduce memory allocation
func (p *DefaultProvider) onComplete(ctx utils.Context, completionType request.OnTestbedCompletionType, testbedName string, clinicURLs map[string]string, actions *mResource.GcAction) *errors.StatusError {
	spec, err := p.getSpec(ctx, testbedName)
	if err != nil {
		return err
	}
	var errs error
	notifier := gc.NewDefaultNotifier()
	var content = fmt.Sprintf("tests on testbed %s: %s", testbedName, completionType)
	if planExecutionID, _ := p.getPlanExecutionID(ctx, testbedName); planExecutionID != "" {
		content = fmt.Sprintf("plan-execution [%s](%s) on testbed %s: %s",
			planExecutionID,
			fmt.Sprintf(rmsConfig.DefaultRMSConfiguration.TCMS.PlanExecutionURLFormat, planExecutionID),
			testbedName, completionType)
	}
	if actions.Artifact != nil {
		content += "\nartifact file list:\n"
		if urls, err := p.onCompleteArchive(ctx.Request().Context(), testbedName, spec.Items, actions.Artifact); err != nil {
			errs = multierror.Append(errs, err)
		} else {
			for objectName, url := range urls {
				content += fmt.Sprintf("\t* [%s](%s)\n", objectName, url.String())
			}
		}

		if len(clinicURLs) > 0 {
			content += "clinic:\n"
			for clusterName, clinicURL := range clinicURLs {
				content += fmt.Sprintf("\t [%s](%s)\n", clusterName, clinicURL)
			}
		}
	}
	if actions.ReserveEnv != nil {
		// use context.Background() because ReserveEnv must be called, otherwise resource wouldn't be reclaimed
		if expireTs, err := p.onCompleteReserveEnv(context.Background(), testbedName, spec.Items, actions.ReserveEnv); err != nil {
			errs = multierror.Append(errs, err)
		} else {
			content += fmt.Sprintf("\nexpire on %s", expireTs.Local().Format(dateFormat))
		}
	}
	if actions.Notification != nil {
		if err := notifier.Notify(ctx.Request().Context(), *actions.Notification, content); err != nil {
			errs = multierror.Append(errs, pkgErrors.Wrapf(err, "notify about %s error", testbedName))
		}
	}
	if errs != nil {
		return errors.NewInternalError(errs)
	}
	return nil
}

func (p *DefaultProvider) onCompleteArchive(ctx context.Context, testbedName string, items []model.TestBedItem, artifact *mResource.GcArtifactAction) (map[string]*url.URL, error) {
	archiver := gc.NewDefaultArchiver(p.newClient)
	var errs error
	var urlMap = make(map[string]*url.URL)

	for _, item := range items {
		iUrls, err := archiver.Archive(ctx, mResource.ResourceHandle{
			Name:    item.Name,
			Testbed: testbedName,
			Type:    item.Type,
		}, *artifact)
		if err != nil {
			errs = multierror.Append(errs, pkgErrors.Wrapf(err, "archive %s.%s error", testbedName, item.Name))
			continue
		}
		for objectName, url := range iUrls {
			urlMap[objectName] = url
		}
	}
	return urlMap, errs
}

func (p *DefaultProvider) onCompleteReserveEnv(ctx context.Context, testbedName string, items []model.TestBedItem, action *mResource.GcReserveEnvAction) (*time.Time, error) {
	var errs error
	reserver := gc.NewDefaultReserver(p.newClient)
	for _, item := range items {
		err := reserver.Reserve(mResource.ResourceHandle{
			Name:    item.Name,
			Testbed: testbedName,
			Type:    item.Type,
		}, *action)
		if err != nil {
			errs = multierror.Append(errs, pkgErrors.Wrapf(err, "reserve %s.%s error", testbedName, item.Name))
		}
	}
	expireTs, err := p.setTestbedWorkingHourExpireTs(ctx, testbedName, action.DurationOnWorkingHour)
	if err != nil {
		errs = multierror.Append(errs, pkgErrors.Wrapf(err, "update expireTs on %s error", testbedName))
	}
	return expireTs, errs
}

func (p *DefaultProvider) CreateResource(_ utils.Context, resource model.TestBedItem) (core.Resource, *errors.StatusError) {
	client, err := p.newClient(resource.Type, resource.Testbed)
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("create resource error: %s", err))
	}
	// construct resource object
	resourceJSON, err := json.Marshal(resource)
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("create resource error: %s", err))
	}
	res, err := client.Create(resourceJSON)
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	return res, nil
}

func (p *DefaultProvider) GetResource(ctx utils.Context, resource mResource.ResourceHandle, withDetail bool) (response.TestbedResponseItem, *errors.StatusError) {
	resp := response.TestbedResponseItem{}
	client, err := p.newClient(resource.Type, resource.Testbed)
	if err != nil {
		return resp, errors.NewInternalError(fmt.Errorf("get resource error: %s", err))
	}
	res, err := client.Get(resource)
	if err != nil {
		return resp, errors.NewInternalError(err)
	}
	resp.ResourceHandle = res.Handle()

	if withDetail {
		endpoints, err := client.GetEndpoints(resource)
		if err != nil {
			ctx.Log().Errorf(err, "%s/%s GetEndpoints failed", resource.Testbed, resource.Name)
			return resp, errors.NewInternalError(err)
		}
		resp.Details.Endpoints = endpoints
	}

	return resp, nil
}

// CreateCredential generate k8s confidential, see https://kubernetes.io/docs/reference/access-authn-authz/authentication/
// for EKS: https://github.com/aws/containers-roadmap/issues/1604#issuecomment-1089585937
func (p *DefaultProvider) CreateCredential(ctx utils.Context, user utils.User, req request.CredentialRequest) (*response.CredentialResponse, *errors.StatusError) {
	privateKey, err := parsePrivateKey(req.PEM)
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	request, err := generateCsr(user.Name, privateKey)
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	// downgrade to use v1beta1, since eks 1.21 has a bug on issue certificate: https://github.com/aws/containers-roadmap/issues/1604#issuecomment-1072660824
	// and we need uses kubernetes.io/legacy-unknown but only v1beta1 API of CSR resource honored this legacy-unknown signer
	rcsr, err := p.ClientSet.CertificatesV1beta1().CertificateSigningRequests().Create(ctx.Request().Context(),
		&v1beta1.CertificateSigningRequest{
			ObjectMeta: metav1.ObjectMeta{
				GenerateName: "rms-",
			},
			Spec: v1beta1.CertificateSigningRequestSpec{
				Request:    request,
				Groups:     []string{"system:authenticated"},
				SignerName: pointer.String(rmsConfig.DefaultRMSConfiguration.K8s.CSRSignerName),
				Usages:     []v1beta1.KeyUsage{v1beta1.UsageClientAuth},
			},
		},
		metav1.CreateOptions{})
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	rcsr.Status.Conditions = append(rcsr.Status.Conditions, v1beta1.CertificateSigningRequestCondition{
		Type:           v1beta1.CertificateApproved,
		Reason:         "auto approved by rms",
		Message:        "auto approved by rms",
		LastUpdateTime: metav1.Now(),
		Status:         corev1.ConditionTrue,
	})
	approval, err := p.ClientSet.CertificatesV1beta1().CertificateSigningRequests().UpdateApproval(ctx.Request().Context(), rcsr, metav1.UpdateOptions{})
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	var signed v1.CertificateSigningRequest

	err = wait.PollUntil(1*time.Second, func() (done bool, err error) {
		csr, err := p.ClientSet.CertificatesV1().CertificateSigningRequests().Get(ctx.Request().Context(), approval.Name, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		done = csr.Status.Certificate != nil && len(csr.Status.Certificate) > 0
		if done {
			signed = *csr
		}
		return
	}, ctx.Request().Context().Done())

	if err != nil {
		return nil, errors.NewInternalError(err)
	}

	token, err := p.GetDefaultServiceAccountTokenOn(ctx, req.TestbedName)
	if err != nil {
		return nil, errors.NewInternalError(pkgErrors.Wrap(err, "get the default sa token error"))
	}
	kubeConfig := generateKubeConfig(*p.clientConfig, signed.Status.Certificate, req.PEM, req.TestbedName, token)
	return &response.CredentialResponse{Credential: kubeConfig, Status: "OK"}, nil
}

func (p *DefaultProvider) ReclaimTestbed() *errors.StatusError {
	nss, err := p.ClientSet.CoreV1().Namespaces().List(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{
			"app.kubernetes.io/managed-by": "test-infra",
		}.AsSelector().String(),
	})
	if err != nil {
		return errors.NewInternalError(err)
	}
	var errs error
	for _, ns := range nss.Items {
		expireTs, _, err := gc.GetLeaseInfoOn(&ns)
		if err == gc.ErrNoLeaseInfoFound {
			continue
		}
		if expireTs.Before(time.Now()) {
			if err := p.ClientSet.CoreV1().Namespaces().Delete(context.TODO(), ns.Name, metav1.DeleteOptions{}); err != nil {
				errs = multierror.Append(errs, err)
			} else {
				log.Log.Infof("testbed %s is deleting", ns.Name)
			}
		}
	}
	if errs != nil {
		return errors.NewInternalError(errs)
	}
	return nil
}

func (p *DefaultProvider) GetLease(ctx utils.Context, testbed string) (*response.LeaseResponse, *errors.StatusError) {
	ns, serr := p.checkNamespace(ctx.Request().Context(), testbed)
	if serr != nil {
		return nil, serr
	}
	expireTs, lease, err := gc.GetLeaseInfoOn(ns)
	if err == gc.ErrNoLeaseInfoFound {
		return nil, nil
	}
	return &response.LeaseResponse{
		Lease:      lease.String(),
		ExpireDate: expireTs.Format(dateFormat),
	}, nil
}

func (p *DefaultProvider) GetGcStrategy(ctx utils.Context, testbed string) (*mResource.GcStrategy, *errors.StatusError) {
	if _, err := p.checkNamespace(ctx.Request().Context(), testbed); err != nil {
		return nil, err
	}
	spec, err := p.getSpec(ctx, testbed)
	if err != nil {
		return nil, err
	}
	return &spec.ReclaimPolicy.GcStrategy, nil
}

func (p *DefaultProvider) RenewLease(ctx utils.Context, testbed string, lease request.DurationStr) (*response.RenewalLeaseResponse, *errors.StatusError) {
	ns, serr := p.checkNamespace(ctx.Request().Context(), testbed)
	if serr != nil {
		return nil, serr
	}
	now := time.Now()

	resp := &response.RenewalLeaseResponse{Name: testbed}
	err := wait.PollImmediate(1*time.Second, 10*time.Second, func() (done bool, err error) {
		newNs := ns.DeepCopy()
		if err = gc.SetLeaseInfoOn(newNs, now, string(lease)); err != nil {
			return false, err
		}
		ns, err = p.ClientSet.CoreV1().Namespaces().Update(ctx.Request().Context(), newNs, metav1.UpdateOptions{})
		if err != nil {
			ctx.Log().Errorf(err, "renew lease error")
			return false, nil
		}
		expireTs, _, err := gc.GetLeaseInfoOn(ns)
		if err != nil {
			return false, err
		}
		resp.ExpireDate = expireTs.Local().Format(dateFormat)
		return true, nil
	})
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	defer func() {
		if err := p.notifyRenewLease(ctx, testbed, resp.ExpireDate); err != nil {
			ctx.Log().Errorf(err, "notify renew on testbed %s error", testbed)
		}
	}()
	return resp, nil
}

func (p *DefaultProvider) notifyRenewLease(ctx utils.Context, testbed string, expireDate string) error {
	s, err := p.GetGcStrategy(ctx, testbed)
	if err != nil {
		return err
	}
	var gcAction = s.OnFailure
	if gcAction == nil {
		gcAction = s.OnCompletion
	}
	if gcAction == nil {
		return nil
	}
	if gcAction.Notification != nil {
		notifier := gc.NewDefaultNotifier()
		return notifier.Notify(ctx.Request().Context(), *gcAction.Notification, fmt.Sprintf("the testbed %s is renewed to %s", testbed, expireDate))
	}
	return nil
}

func (p *DefaultProvider) GetDefaultServiceAccountTokenOn(ctx utils.Context, ns string) (token string, err error) {
	sa, err := p.ClientSet.CoreV1().ServiceAccounts(ns).Get(ctx.Request().Context(), "default", metav1.GetOptions{})
	if err != nil {
		return "", err
	}
	if len(sa.Secrets) == 0 {
		return "", fmt.Errorf("unknown secret name of sa %s", sa.Name)
	}
	secretName := sa.Secrets[0].Name
	secret, err := p.ClientSet.CoreV1().Secrets(ns).Get(ctx.Request().Context(), secretName, metav1.GetOptions{})
	if err != nil {
		return "", err
	}
	d := secret.Data["token"]
	return string(d), nil
}

func (p *DefaultProvider) checkQuotaLimitOfUser(ctx utils.Context, user *utils.User) *errors.StatusError {
	quota := rmsConfig.DefaultRMSConfiguration.ResourceQuota
	if user.IsAdmin() {
		return nil
	}
	// check the testbed quota limit
	accounting := generateAccountingNameFor(user)
	nss, err := p.ClientSet.CoreV1().Namespaces().List(ctx.Request().Context(), metav1.ListOptions{
		LabelSelector: labels.Set{
			"app.kubernetes.io/managed-by": "test-infra",
			"accounting":                   accounting,
		}.AsSelector().String(),
	})
	if err != nil {
		return errors.NewInternalError(err)
	}
	if len(nss.Items) >= int(quota.TestbedPerUser) {
		var testbedList []string
		for _, ns := range nss.Items {
			testbedList = append(testbedList, ns.Name)
		}
		return errors.NewTooManyRequest(fmt.Sprintf("reach your testbed quota limit(%d), please check your testbed list and delete them first: %s", quota.TestbedPerUser, strings.Join(testbedList, ",")))
	}
	return nil
}

func (p *DefaultProvider) getSpec(_ utils.Context, testbed string) (*request.CreateTestbedRequest, *errors.StatusError) {
	testbedClient := k8s.NewTestbedClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	resource, err := testbedClient.Get(mResource.ResourceHandle{
		Name: testbed,
		Type: mResource.ResourceTypeTestbed,
	})
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	tb := resource.(*k8s.Testbed)
	spec, err := tb.GetSpec()
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	return spec, nil
}

func (p *DefaultProvider) getPlanExecutionID(ctx utils.Context, testbed string) (string, *errors.StatusError) {
	ns, err := p.ClientSet.CoreV1().Namespaces().Get(ctx.Request().Context(), testbed, metav1.GetOptions{})
	if err != nil {
		return "", errors.NewInternalError(err)
	}
	accounting := ns.Labels["accounting"]
	if !strings.HasPrefix(accounting, "ID-") {
		return "", nil
	}
	return strings.TrimPrefix(accounting, "ID-"), nil
}

//nolint:gocyclo
func (p *DefaultProvider) newClient(kind mResource.ResourceType, testbed string) (sdkResouce.Client, error) {
	var client sdkResouce.Client
	switch kind {
	case mResource.ResourceTypeTiDBCluster:
		client = k8s.NewTiDBClusterClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	case mResource.ResourceTypeWorkloadNode:
		client = k8s.NewWorkloadNodeClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	case mResource.ResourceTypeTiDBLightning:
		client = k8s.NewLightningClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	case mResource.ResourceTypeDMCluster:
		client = k8s.NewDMClusterClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	case mResource.ResourceTypeKafkaCluster:
		client = k8s.NewKafkaClusterClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	case mResource.ResourceTypeBRNode:
		client = k8s.NewBRClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	case mResource.ResourceTypeMySQLNode:
		client = k8s.NewMySQLClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	case mResource.ResourceTypeDrainer:
		client = k8s.NewTiDBDrainerClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	case mResource.ResourceTypeLoadBalancer:
		client = k8s.NewLoadBalancerClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient))
	case mResource.ResourceTypeTiDBCloudCluster:
		var err error
		client, err = k8s.NewTiDBCloudClusterClient(k8s.NewKubeClient(testbed, p.clientConfig, p.kubeConfig, p.ClientSet, p.DynamicClient), &rmsConfig.DefaultRMSConfiguration.TiDBCloudConfiguration)
		if err != nil {
			return nil, fmt.Errorf("create client %s error: %s", kind, err)
		}
	default:
		return nil, fmt.Errorf("create resource error, unknown resource type: %s", kind)
	}
	return client, nil
}

func (p *DefaultProvider) getCurrentTestbeds() ([]corev1.Namespace, error) {
	nss, err := p.ClientSet.CoreV1().Namespaces().List(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{
			"app.kubernetes.io/managed-by": "test-infra",
		}.AsSelector().String(),
	})
	if err != nil {
		return nil, err
	}
	return nss.Items, nil
}

func (p *DefaultProvider) setTestbedWorkingHourExpireTs(ctx context.Context, testbedName string, durationOnWorkingHour string) (*time.Time, *errors.StatusError) {
	duration, err := time.ParseDuration(durationOnWorkingHour)
	if err != nil {
		return nil, errors.NewBadRequest(err.Error())
	}
	var expireTs time.Time
	var ns *corev1.Namespace
	err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		ns, err = p.ClientSet.CoreV1().Namespaces().Get(ctx, testbedName, metav1.GetOptions{})
		if err != nil {
			return err
		}
		expireTs = calcExpireTsOnWorkingHour(duration)
		if err = gc.SetExpireTsOn(ns, expireTs, durationOnWorkingHour); err != nil {
			return err
		}
		_, updateErr := p.ClientSet.CoreV1().Namespaces().Update(ctx, ns, metav1.UpdateOptions{})
		return updateErr
	})
	if err != nil {
		return nil, errors.NewInternalError(err)
	}
	return &expireTs, nil
}

func (p *DefaultProvider) checkNamespace(ctx context.Context, testbed string) (*corev1.Namespace, *errors.StatusError) {
	ns, err := p.ClientSet.CoreV1().Namespaces().Get(ctx, testbed, metav1.GetOptions{})
	if err != nil {
		if kerrors.IsNotFound(err) {
			return nil, errors.NewNotFound(fmt.Sprintf("namespace %s", testbed))
		}
		return nil, errors.NewInternalError(err)
	} else if ns.Status.Phase != corev1.NamespaceActive {
		return nil, errors.NewNotFound(testbed)
	}
	return ns, nil
}

func calcExpireTsOnWorkingHour(duration time.Duration) time.Time {
	if duration == 0 {
		// set now as the expireTs
		return time.Now()
	}
	// otherwise, put the expireTs on the current working hour or next working hour if now time is beyond today's working hour
	var workingHours = []int{10, 19}
	now := time.Now().Local()
	hour := now.Hour()
	// when the switch `NoReserveEnvWhenNotInWorkingHour` is on, don't reserve.
	if rmsConfig.DefaultRMSConfiguration.ReclaimConfiguration.NoReserveEnvWhenNotInWorkingHour {
		if hour < workingHours[0] || workingHours[1] < hour {
			return time.Now()
		}
	}
	var workingDate = now
	if hour < workingHours[0] {
		workingDate = time.Date(now.Year(), now.Month(), now.Day(), workingHours[0], 0, 0, 0, time.Local)
	} else if workingHours[1] <= hour {
		nextDay := now.Add(time.Hour * 24)
		workingDate = time.Date(nextDay.Year(), nextDay.Month(), nextDay.Day(), workingHours[0], 0, 0, 0, time.Local)
	}
	return workingDate.Add(duration)
}

func parsePrivateKey(pk []byte) (privateKey interface{}, err error) {
	block, _ := pem.Decode(pk)
	if block == nil || !strings.Contains(block.Type, "PRIVATE KEY") {
		return nil, errors.NewBadRequest("not a PRIVATE KEY")
	}
	if privateKey, err = x509.ParsePKCS8PrivateKey(block.Bytes); err != nil {
		privateKey, err = x509.ParsePKCS1PrivateKey(block.Bytes)
	}
	return
}

func generateKubeConfig(config rest.Config, cert []byte, key []byte, ns string, token string) []byte {
	template := `apiVersion: v1
kind: Config
clusters:
- name: default-cluster
  cluster:
    certificate-authority-data: ${ca}
    server: ${server}
contexts:
- name: default-context
  context:
    cluster: default-cluster
    namespace: ${namespace}
    user: default-user
current-context: default-context
users:
- name: default-user
  user:
    client-certificate-data: ${cert}
    client-key-data: ${key}
    token: ${token}
`
	server := rmsConfig.DefaultRMSConfiguration.K8s.K8sApiserver
	if server == "" {
		server = config.Host
	}
	ca, _ := dataFromSliceOrFile(config.CAData, config.CAFile)
	t := strings.Replace(template, "${ca}", base64.StdEncoding.EncodeToString(ca), 1)
	t = strings.Replace(t, "${server}", server, 1)
	t = strings.Replace(t, "${cert}", base64.StdEncoding.EncodeToString(cert), 1)
	t = strings.Replace(t, "${key}", base64.StdEncoding.EncodeToString(key), 1)
	t = strings.Replace(t, "${namespace}", ns, 1)
	t = strings.Replace(t, "${token}", token, 1)
	return []byte(t)
}

func dataFromSliceOrFile(data []byte, file string) ([]byte, error) {
	if len(data) > 0 {
		return data, nil
	}
	if len(file) > 0 {
		fileData, err := ioutil.ReadFile(file)
		if err != nil {
			return []byte{}, err
		}
		return fileData, nil
	}
	return nil, nil
}

func generateCsr(name string, key interface{}) ([]byte, error) {
	var signer crypto.Signer
	var pub x509.PublicKeyAlgorithm
	var sig x509.SignatureAlgorithm
	switch key.(type) {
	case *rsa.PrivateKey:
		signer = key.(crypto.Signer)
		pub = x509.RSA
		sig = x509.SHA256WithRSA
	case *ecdsa.PrivateKey:
		signer = key.(crypto.Signer)
		pub = x509.ECDSA
		sig = x509.ECDSAWithSHA256
	default:
		return nil, errors.NewBadRequest("unknown private key")
	}
	certificateRequest := x509.CertificateRequest{
		Subject:            pkix.Name{CommonName: name},
		SignatureAlgorithm: sig,
		PublicKey:          signer.Public(),
		PublicKeyAlgorithm: pub,
	}
	csrDer, err := x509.CreateCertificateRequest(rand.Reader, &certificateRequest, signer)
	if err != nil {
		return nil, err
	}
	csr := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: csrDer})
	return csr, nil
}