package k8s

import (
	"context"
	"encoding/json"
	"fmt"
	"net"
	"net/url"
	"os"
	"path"
	"strings"

	"github.com/creasty/defaults"
	"github.com/hashicorp/go-multierror"
	"github.com/pingcap/errors"
	"go.uber.org/zap"
	v1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/utils/pointer"

	apiresource "github.com/pingcap/test-infra/common/model/resource"
	"github.com/pingcap/test-infra/sdk/core"
	sdkresource "github.com/pingcap/test-infra/sdk/resource"
)

const (
	systemdReservedMemoryBytes = 512 * 1024 * 1024 // 512 MiB
	systemdConfigMapKey        = "systemd"
)

var (
	_ sdkresource.Client = WorkloadNodeClient{}
)

type WorkloadNodeClient struct {
	*KubeClientSet
}

func NewWorkloadNodeClient(k *KubeClientSet) WorkloadNodeClient {
	return WorkloadNodeClient{k}
}

func (t WorkloadNodeClient) Create(spec json.RawMessage) (core.Resource, error) {
	cm, sts, service, loadBalanceService, apiRes, err := t.buildWorkloadNodeObject(spec)
	if err != nil {
		return nil, err
	}
	if apiRes.Qos != nil {
		if err := t.applyQos(apiRes, sts); err != nil {
			return nil, err
		}
	}

	if cm != nil {
		if cm, err = t.ClientSet.CoreV1().ConfigMaps(t.Namespace).Create(context.TODO(), cm, metav1.CreateOptions{}); err != nil {
			return nil, err
		}
	}

	service, err = t.ClientSet.CoreV1().Services(t.Namespace).Create(context.TODO(), service, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}
	loadBalanceService, err = t.ClientSet.CoreV1().Services(t.Namespace).Create(context.TODO(), loadBalanceService, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}
	sts, err = t.ClientSet.AppsV1().StatefulSets(t.Namespace).Create(context.TODO(), sts, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}
	resource := &WorkloadNode{
		KubeClientSet:      *t.KubeClientSet,
		WorkloadNode:       *apiRes,
		cm:                 cm,
		sts:                sts,
		service:            service,
		loadBalanceService: loadBalanceService,
	}
	resource.Provider = t.FQDN()
	if err := resource.Sync(); err != nil {
		return nil, err
	}

	err = t.KubeClientSet.AddResourceSpecToConfigMap(resource.ResourceHandle, resource.Testbed, spec)
	if err != nil {
		return nil, err
	}

	return resource, nil
}

func (t WorkloadNodeClient) Get(handle apiresource.ResourceHandle) (core.Resource, error) {
	resource := &WorkloadNode{
		KubeClientSet: *t.KubeClientSet,
		WorkloadNode: apiresource.WorkloadNode{
			ResourceHandle: handle,
		},
	}
	resource.Provider = t.FQDN()
	if err := resource.Sync(); err != nil {
		return nil, err
	}
	return resource, nil
}

func (t WorkloadNodeClient) GetEndpoints(handle apiresource.ResourceHandle) (apiresource.ResourceEndpoints, error) {
	res := apiresource.ResourceEndpoints{}

	svcList, err := t.ClientSet.CoreV1().
		Services(handle.Testbed).
		List(context.TODO(), metav1.ListOptions{
			LabelSelector: labels.Set{"test-infra/name": handle.Name}.AsSelector().String(),
		})
	if err != nil {
		return res, err
	}

	for _, svc := range svcList.Items {
		if svc.Spec.Type != corev1.ServiceTypeNodePort {
			continue
		}
		endPoint := apiresource.Endpoint{
			Type: apiresource.EndpointTypeOther,
			Name: svc.Name,
		}

		pods, err := t.ClientSet.CoreV1().Pods(handle.Testbed).List(context.TODO(), metav1.ListOptions{
			LabelSelector: labels.FormatLabels(svc.Spec.Selector),
		})
		if err != nil {
			return res, err
		}
		if len(pods.Items) == 0 {
			return res, fmt.Errorf("service %s/%s has no target pod", svc.Namespace, svc.Name)
		}

		// No protocol here
		endPoint.ExternalSvcAddress = net.JoinHostPort(pods.Items[0].Status.HostIP, fmt.Sprintf("%d", svc.Spec.Ports[0].NodePort))

		res = append(res, endPoint)
		break
	}

	return res, nil
}

func (t WorkloadNodeClient) Delete(handle apiresource.ResourceHandle) error {
	var errs error
	if err := t.ClientSet.AppsV1().StatefulSets(t.Namespace).Delete(context.TODO(), handle.Name, metav1.DeleteOptions{
		PropagationPolicy: &[]metav1.DeletionPropagation{metav1.DeletePropagationForeground}[0],
	}); err != nil {
		errs = multierror.Append(errs, err)
	}

	if err := t.ClientSet.CoreV1().Services(t.Namespace).Delete(context.TODO(), handle.Name+"-peer", metav1.DeleteOptions{}); err != nil {
		errs = multierror.Append(errs, err)
	}
	if err := t.ClientSet.CoreV1().Services(t.Namespace).Delete(context.TODO(), handle.Name+"-lb", metav1.DeleteOptions{}); err != nil {
		errs = multierror.Append(errs, err)
	}

	if err := t.ClientSet.CoreV1().ConfigMaps(t.Namespace).DeleteCollection(context.TODO(), metav1.DeleteOptions{}, metav1.ListOptions{
		LabelSelector: labels.Set{TestInfraName: handle.Name}.AsSelector().String(),
	}); err != nil {
		errs = multierror.Append(errs, err)
	}

	if err := t.ClientSet.CoreV1().PersistentVolumeClaims(t.Namespace).DeleteCollection(context.TODO(), metav1.DeleteOptions{}, metav1.ListOptions{
		LabelSelector: labels.Set{TestInfraName: handle.Name}.AsSelector().String(),
	}); err != nil {
		errs = multierror.Append(errs, err)
	}

	if err := t.KubeClientSet.RemoveResourceSpecFromConfigMap(handle, handle.Testbed); err != nil {
		errs = multierror.Append(errs, err)
	}

	return errs
}

func (t WorkloadNodeClient) ReserveEnv(handle apiresource.ResourceHandle, action apiresource.GcReserveEnvAction) error {
	return nil
}

func (t WorkloadNodeClient) Artifact(handle apiresource.ResourceHandle, action apiresource.GcArtifactAction, archiver sdkresource.ArchiverFunc) (map[string]*url.URL, error) {
	resource, err := t.Get(handle)
	if err != nil {
		return nil, err
	}
	workloadNode := resource.(*WorkloadNode)
	return artifactWorkloadNode(t.KubeClientSet, workloadNode, archiver)
}

func artifactWorkloadNode(t *KubeClientSet, workloadNode *WorkloadNode, archiver sdkresource.ArchiverFunc) (map[string]*url.URL, error) {
	pods, err := t.ClientSet.CoreV1().Pods(t.Namespace).List(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{TestInfraName: workloadNode.Name}.AsSelector().String()})
	if err != nil {
		return nil, err
	}
	if len(pods.Items) != int(workloadNode.Spec.Replicas) {
		return nil, fmt.Errorf("expect %d pod in %s, got %d", workloadNode.Spec.Replicas, workloadNode.Name, len(pods.Items))
	}

	var podName string
	var urlMap = make(map[string]*url.URL)
	var errs error

	for i := range pods.Items {
		podName = pods.Items[i].Name
		for _, artifactPath := range workloadNode.Spec.Artifact.ArtifactPaths {
			artifactPath = path.Clean(artifactPath)
			tarballName := podName + "/" + strings.ReplaceAll(artifactPath[1:], "/", ".") + ".tar"

			tarStream := t.TarStreamFrom(podName, workloadNode.Spec.Container.Name, artifactPath)
			defer tarStream.Close()

			url, err := archiver(tarballName, tarStream)
			if err != nil {
				errs = multierror.Append(errs, err)
				continue
			}
			urlMap[tarballName] = url
		}
	}
	if errs != nil {
		sugar.Errorw("artifact workload error", zap.String("namespace", t.Namespace), zap.String("resource", workloadNode.Name), zap.Error(errs))
	}
	return urlMap, nil
}

func (t WorkloadNodeClient) FQDN() string {
	return sdkresource.FQDN(t)
}

// configMap can be used as `filesystem`, `env`, `command-line args` etc.
func (t WorkloadNodeClient) buildConfigMapObj(resHandle *apiresource.WorkloadNode) (cm *corev1.ConfigMap) {
	// use as env
	data := make(map[string]string)
	// copy container env to the data map because the container template uses valueFrom form on the env field
	for key, value := range resHandle.Spec.Container.Env {
		data[key] = value
	}
	// use as systemd memory limit injection
	data[systemdConfigMapKey] = ""
	for resourceType, resourceQuantity := range resHandle.Spec.Container.Resources.Requests {
		if resourceType != apiresource.Memory {
			continue
		}

		quantity, err := resource.ParseQuantity(string(resourceQuantity))
		if err == nil &&
			quantity.Value()-systemdReservedMemoryBytes > 0 {
			// set system.slice MemoryLimit to requests.memory - systemdReservedMemoryBytes
			data[systemdConfigMapKey] = fmt.Sprintf("[Slice]\nMemoryLimit=%d\n", quantity.Value()-systemdReservedMemoryBytes)
		}
	}

	return &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      resHandle.Name + "-config",
			Namespace: t.Namespace,
			Labels: map[string]string{
				"app.kubernetes.io/managed-by": "test-infra",
				TestInfraName:                  resHandle.Name,
			},
		},
		Data: data,
	}
}

func (t WorkloadNodeClient) buildProbe(apiprobe *apiresource.Probe) (probe *corev1.Probe) {
	if apiprobe == nil {
		return nil
	}
	return &corev1.Probe{
		ProbeHandler: corev1.ProbeHandler{
			Exec: &corev1.ExecAction{
				Command: apiprobe.Exec.Command,
			},
		},
		InitialDelaySeconds:           apiprobe.InitialDelaySeconds,
		TimeoutSeconds:                apiprobe.TimeoutSeconds,
		PeriodSeconds:                 apiprobe.PeriodSeconds,
		SuccessThreshold:              apiprobe.SuccessThreshold,
		FailureThreshold:              apiprobe.FailureThreshold,
		TerminationGracePeriodSeconds: apiprobe.TerminationGracePeriodSeconds,
	}
}

func (t WorkloadNodeClient) buildWorkloadNodeObject(params json.RawMessage) (cm *corev1.ConfigMap, sts *v1.StatefulSet,
	service *corev1.Service, loadBalanceService *corev1.Service, resHandle *apiresource.WorkloadNode, err error) {
	resHandle = &apiresource.WorkloadNode{}

	// TODO: add validation on WorkloadNode
	if err := json.Unmarshal(params, resHandle); err != nil {
		return nil, nil, nil, nil, nil, errors.Trace(err)
	}

	if err := defaults.Set(resHandle); err != nil {
		return nil, nil, nil, nil, nil, errors.Trace(err)
	}

	owner, err := GetOwnerReferenceFromTestbed(t.KubeClientSet, resHandle.Testbed)
	if err != nil {
		return nil, nil, nil, nil, nil, err
	}

	cm = t.buildConfigMapObj(resHandle)
	app, serviceName, port := resHandle.Name, resHandle.Name+"-peer", resHandle.Spec.Container.Port

	pod := corev1.PodSpec{
		NodeSelector: resHandle.Spec.Container.NodeSelector,
		// tolerate test-infra dedicated nodes
		Tolerations: []corev1.Toleration{{
			Key:      "test-infra",
			Operator: "Equal",
			Value:    "dedicated",
			Effect:   "NoSchedule",
		}},
		Containers: []corev1.Container{{
			Name:            resHandle.Spec.Container.Name,
			Image:           resHandle.Spec.Container.Image,
			Command:         resHandle.Spec.Container.Command,
			ImagePullPolicy: corev1.PullPolicy(resHandle.Spec.Container.ImagePullPolicy),
			Env:             t.buildPodEnv(resHandle),
			Args:            resHandle.Spec.Container.Args,
			WorkingDir:      resHandle.Spec.Container.WorkingDir,
			Resources: corev1.ResourceRequirements{
				Limits:   make(corev1.ResourceList),
				Requests: make(corev1.ResourceList),
			},
			LivenessProbe:  t.buildProbe(resHandle.Spec.Container.LivenessProbe),
			ReadinessProbe: t.buildProbe(resHandle.Spec.Container.ReadinessProbe),
			VolumeMounts: []corev1.VolumeMount{
				// inject memory limit for systemd workloadNode. takes no effect to non-systemd ones
				{Name: "systemd", MountPath: "/etc/systemd/system/system.slice.d"},
			},
		}},
		Volumes: []corev1.Volume{
			{
				Name: "systemd",
				VolumeSource: corev1.VolumeSource{
					ConfigMap: &corev1.ConfigMapVolumeSource{
						LocalObjectReference: corev1.LocalObjectReference{
							Name: cm.Name,
						},
						Items: []corev1.KeyToPath{
							{Key: systemdConfigMapKey, Path: "test-infra.MemoryLimit.conf"},
						},
					},
				},
			},
		},
	}
	if resHandle.Spec.Container.CGroupMountMode != nil {
		switch *resHandle.Spec.Container.CGroupMountMode {
		case "readonly":
			pod.Volumes = append(pod.Volumes, corev1.Volume{
				Name: "cgroup",
				VolumeSource: corev1.VolumeSource{
					HostPath: &corev1.HostPathVolumeSource{
						Path: "/sys/fs/cgroup",
						Type: (*corev1.HostPathType)(pointer.String(string(corev1.HostPathDirectory))),
					},
				},
			})
			pod.Containers[0].VolumeMounts = append(pod.Containers[0].VolumeMounts, corev1.VolumeMount{
				Name:      "cgroup",
				ReadOnly:  true,
				MountPath: "/sys/fs/cgroup",
			})
		}
	}
	// set SecurityContext, only support add Capability now
	if resHandle.Spec.Container.SecurityContext != nil {
		var caps []corev1.Capability
		for _, cap := range resHandle.Spec.Container.SecurityContext.Capabilities.Add {
			caps = append(caps, corev1.Capability(cap))
		}
		pod.Containers[0].SecurityContext = &corev1.SecurityContext{Capabilities: &corev1.Capabilities{
			Add: caps,
		}}
	}

	if port != 0 {
		pod.Containers[0].Ports = []corev1.ContainerPort{
			{
				Name:          app,
				ContainerPort: port,
			},
		}
	}

	podLabels := map[string]string{
		"app.kubernetes.io/managed-by": "test-infra",
		TestInfraName:                  resHandle.Name,
	}
	for k, v := range resHandle.Spec.Container.Labels {
		podLabels[k] = v
	}

	sts = &v1.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:      resHandle.Name,
			Namespace: t.Namespace,
			Labels: map[string]string{
				"app.kubernetes.io/managed-by": "test-infra",
				TestInfraName:                  resHandle.Name,
			},
			OwnerReferences: []metav1.OwnerReference{*owner},
		},
		Spec: v1.StatefulSetSpec{
			Replicas: &resHandle.Spec.Replicas,
			Selector: &metav1.LabelSelector{
				// https://kubernetes.io/docs/concepts/overview/working-with-objects/common-labels/
				MatchLabels: map[string]string{
					"app.kubernetes.io/managed-by": "test-infra",
					TestInfraName:                  resHandle.Name,
				},
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: podLabels,
				},
				Spec: pod,
			},
			ServiceName: serviceName,
		},
	}

	// apply resource requests & limits
	for res, request := range resHandle.Spec.Container.Resources.Requests {
		switch res {
		case apiresource.CPU, apiresource.Memory:
			quantity, err := resource.ParseQuantity(string(request))
			if err != nil {
				return nil, nil, nil, nil, nil, errors.Annotatef(err, "parse quantity on %s", request)
			}
			pod.Containers[0].Resources.Requests[corev1.ResourceName(res)] = quantity
			pod.Containers[0].Resources.Limits[corev1.ResourceName(res)] = quantity
		case apiresource.Storage:
			quantity, err := resource.ParseQuantity(string(request))
			if err != nil {
				return nil, nil, nil, nil, nil, errors.Annotatef(err, "parse quantity on %s", request)
			}
			pod.Containers[0].VolumeMounts = append(pod.Containers[0].VolumeMounts, corev1.VolumeMount{
				Name:      "data",
				MountPath: resHandle.Spec.Container.VolumeMount.MountPath,
			})
			sts.Spec.VolumeClaimTemplates = append(sts.Spec.VolumeClaimTemplates, corev1.PersistentVolumeClaim{
				ObjectMeta: metav1.ObjectMeta{
					Name: "data",
				},
				Spec: corev1.PersistentVolumeClaimSpec{
					Resources: corev1.ResourceRequirements{
						Requests: corev1.ResourceList{
							corev1.ResourceStorage: quantity,
						},
					},
					AccessModes:      []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
					StorageClassName: resHandle.Spec.Container.StorageClassName,
				},
			})
		}
	}
	// only CPU & memory limits are handled
	// Without explicit configuration, limit will be set to request(for historical reason)
	for res, limit := range resHandle.Spec.Container.Resources.Limits {
		switch res {
		case apiresource.CPU, apiresource.Memory:
			quantity, err := resource.ParseQuantity(string(limit))
			if err != nil {
				return nil, nil, nil, nil, nil, errors.Annotatef(err, "parse quantity on %s", limit)
			}
			pod.Containers[0].Resources.Limits[corev1.ResourceName(res)] = quantity
		}
	}

	err = t.applyAdditionalVolumes(resHandle, sts)
	if err != nil {
		return nil, nil, nil, nil, nil, errors.Trace(err)
	}

	// headless service for Statefulset
	service = t.buildService(resHandle.Name, serviceName, corev1.ClusterIPNone, corev1.ServiceTypeClusterIP, owner, port)
	// clusterIP service as loadbalancer
	loadBalanceService = t.buildService(resHandle.Name, resHandle.Name+"-lb", "", corev1.ServiceType(resHandle.Spec.LoadBalancer.ServiceType), owner, resHandle.Spec.LoadBalancer.ContainerPort)

	return cm, sts, service, loadBalanceService, resHandle, nil
}

func (t WorkloadNodeClient) buildService(resHandleName, serviceName, ip string, serviceType corev1.ServiceType,
	owner *metav1.OwnerReference, port int32) *corev1.Service {

	service := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      serviceName,
			Namespace: t.Namespace,
			Labels: map[string]string{
				"app.kubernetes.io/managed-by": "test-infra",
				TestInfraName:                  resHandleName,
			},
			OwnerReferences: []metav1.OwnerReference{*owner},
		},
		// if ClusterIP is `None`, Type must set to `ClusterIP`, else no need to set `ClusterIP`
		Spec: corev1.ServiceSpec{
			Type:      serviceType,
			ClusterIP: ip,
			Selector: map[string]string{
				"app.kubernetes.io/managed-by": "test-infra",
				TestInfraName:                  resHandleName,
			},
		},
	}

	if port != 0 {
		service.Spec.Ports = []corev1.ServicePort{
			{
				Name: resHandleName,
				Port: port,
			},
		}
	}

	return service
}

// initialize env from configMap, for `{p}`, the corresponding configMap has name `{p}-config`
func (t WorkloadNodeClient) buildPodEnv(resHandle *apiresource.WorkloadNode) []corev1.EnvVar {
	env := make([]corev1.EnvVar, 0)
	for k := range resHandle.Spec.Container.Env {
		env = append(env, corev1.EnvVar{
			Name: k,
			ValueFrom: &corev1.EnvVarSource{
				ConfigMapKeyRef: &corev1.ConfigMapKeySelector{
					LocalObjectReference: corev1.LocalObjectReference{Name: resHandle.Name + "-config"},
					Key:                  k,
				},
			},
		})
	}
	return env
}

// applyAdditionalVolumes inplace modifies to mount some extra volumes
func (t WorkloadNodeClient) applyAdditionalVolumes(res *apiresource.WorkloadNode, sts *v1.StatefulSet) error {
	if len(res.Spec.Container.AdditionalVolumes) == 0 {
		return nil
	}

	// 1. clean up pv.spec.claimRef, otherwise the pv cannot be Available
	for _, vol := range res.Spec.Container.AdditionalVolumes {
		if vol.VolumeSource.PersistentVolume == nil {
			continue
		}

		pv, err := t.ClientSet.CoreV1().PersistentVolumes().Get(context.TODO(),
			vol.VolumeSource.PersistentVolume.PersistentVolumeName, metav1.GetOptions{})
		if err != nil {
			return errors.Annotatef(err, "get pv %s", vol.VolumeSource.PersistentVolume.PersistentVolumeName)
		}

		if pv.Spec.ClaimRef != nil {
			_, err = t.ClientSet.CoreV1().PersistentVolumeClaims(pv.Spec.ClaimRef.Namespace).Get(context.TODO(), pv.Spec.ClaimRef.Name, metav1.GetOptions{})
			switch {
			case apierrors.IsNotFound(err):
				// pvc not exist anymore, safe to cleanup claimRef
			case err != nil:
				return errors.Annotatef(err, "get pvc %s/%s", pv.Spec.ClaimRef.Namespace, pv.Spec.ClaimRef.Name)
			case err == nil:
				return errors.Errorf("pv %s is still binded to pvc %s/%s, cannot mount to %s/%s", pv.Name, pv.Spec.ClaimRef.Namespace, pv.Spec.ClaimRef.Name, res.Testbed, res.Name)
			}
		}

		pv.Spec.ClaimRef = nil
		pv.Status.Phase = corev1.VolumeAvailable
		_, err = t.ClientSet.CoreV1().PersistentVolumes().Update(context.TODO(), pv, metav1.UpdateOptions{})
		if err != nil {
			return errors.Annotatef(err, "update pv %s", vol.VolumeSource.PersistentVolume.PersistentVolumeName)
		}
	}

	// 2. add additional volume to statefulset
	for _, vol := range res.Spec.Container.AdditionalVolumes {
		if vol.VolumeSource.PersistentVolume == nil {
			continue
		}

		volumeName := fmt.Sprintf("%s-%s", res.Name, vol.VolumeSource.PersistentVolume.PersistentVolumeName)
		sts.Spec.VolumeClaimTemplates = append(sts.Spec.VolumeClaimTemplates, corev1.PersistentVolumeClaim{
			ObjectMeta: metav1.ObjectMeta{
				Name: volumeName,
			},
			Spec: corev1.PersistentVolumeClaimSpec{
				Resources: corev1.ResourceRequirements{
					Requests: corev1.ResourceList{
						corev1.ResourceStorage: resource.MustParse("1Gi"),
					},
				},
				AccessModes:      []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
				StorageClassName: &vol.VolumeSource.PersistentVolume.StorageClassName,
				VolumeName:       vol.VolumeSource.PersistentVolume.PersistentVolumeName, // bind to specific pv
			},
		})

		sts.Spec.Template.Spec.Containers[0].VolumeMounts = append(sts.Spec.Template.Spec.Containers[0].VolumeMounts, corev1.VolumeMount{
			Name:      volumeName,
			MountPath: vol.VolumeMountPath,
		})
	}

	return nil
}

func (t WorkloadNodeClient) applyQos(res *apiresource.WorkloadNode, sts *v1.StatefulSet) error {
	qos := res.Qos
	pg, err := t.DynamicClient.Resource(podGroupGVR).Namespace(t.Namespace).Get(context.TODO(), qos.PodGroup, metav1.GetOptions{})
	if err != nil {
		return errors.Trace(err)
	}
	subGroupSpec := map[string]interface{}{
		"minMember": int64(1),
	}
	if res.Qos.Exclusive != nil {
		subGroupSpec["exclusive"] = *res.Qos.Exclusive
	}
	if sts.Spec.Template.ObjectMeta.Labels == nil {
		sts.Spec.Template.ObjectMeta.Labels = make(map[string]string)
	}
	sts.Spec.Template.ObjectMeta.Labels[podGroupLabelKey] = fmt.Sprintf("%s.%s", qos.PodGroup, res.Name)

	// append a subgroup spec on the podgroup: $name_$comp format, for example: tc-1_tidb
	_ = unstructured.SetNestedField(pg.Object, subGroupSpec, "spec", "subGroups", res.Name)
	if _, err := t.DynamicClient.Resource(podGroupGVR).Namespace(t.Namespace).Update(context.TODO(), pg, metav1.UpdateOptions{}); err != nil {
		return errors.Annotatef(err, "update podGroup %s", pg.GetName())
	}
	return nil
}

type workloadNodeGetter struct{ handle apiresource.ResourceHandle }

func (t *workloadNodeGetter) Get() (core.Resource, error) {
	k, err := NewKubeClientFromKubeConfig(t.handle.Testbed, os.Getenv("KUBECONFIG"))
	if err != nil {
		return nil, err
	}
	return NewWorkloadNodeClient(k).Get(t.handle)
}

func init() {
	core.RegisterProvider(sdkresource.FQDN(WorkloadNodeClient{}),
		func(handle apiresource.ResourceHandle) core.ResourceGetter {
			return &workloadNodeGetter{handle: handle}
		})
}