package audit

import (
	"context"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/Masterminds/semver"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	apps "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	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/types"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	sdk "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

var auditLogVersionConstraint, _ = semver.NewConstraint(">= v5.4.0")
var auditSidecarDynamicLogPathVersionConstraint, _ = semver.NewConstraint(">= v1.0.0")

const (
	AuditLogName   = "auditlog"
	MountPath      = "/var/log/auditlog"
	configFilePath = "/etc/audit/audit.yaml"

	pluginName = "audit-1"

	// Note the slowlog file(--log-slow-query) parameter of tidb will be filepath.Join(MountPath, SlowlogVolumeName)
	// see https://github.com/pingcap/tidb-operator/issues/4396 for detail.
	SlowlogVolumeName = "slowlog"

	MetricsPortAnnotationKey = "tidb.auditlog.prometheus.io/port"
	MetricsPort              = 9090

	GCPSecretName            = "db-audit"
	GCPCredentialPath        = "/.gcp/credentials.json"
	gcpCredentialKey         = "privateKey"
	GCPCredentialsVolumeName = "audit-credentials"

	DefaultStorageSize = "30Gi"
	StorageClassName   = "auditlog"
	// HyperdiskBalancedStorageClassName is the StorageClass name used for N4 machine type,
	// which requires the hyperdisk-balanced storage type.
	HyperdiskBalancedStorageClassName = "auditlog-hyperdisk-balanced"
)

// MachineTypeToAuditLogSC maps GCP machine type prefixes to the corresponding
// StorageClass to be used for audit log volumes. Currently only n4-series nodes use HyperdiskBalancedStorageClass.
var MachineTypeToAuditLogSC = map[string]string{
	// GCP:
	// n4-standard-2
	// c4a-standard-2
	"n4-":  HyperdiskBalancedStorageClassName,
	"c4a-": HyperdiskBalancedStorageClassName,
}

func GetStorageClassName(instanceType string) string {
	storageClass := StorageClassName // default storage class

	// Check if instanceType has a special storage class mapping
	for prefix, sc := range MachineTypeToAuditLogSC {
		if strings.HasPrefix(instanceType, prefix) {
			storageClass = sc
			break
		}
	}
	return storageClass
}

func buildAuditLogStorageVolume(audit *cloudv1.Audit, instanceType string) v1alpha1.StorageVolume {
	return v1alpha1.StorageVolume{
		Name:             SlowlogVolumeName,
		StorageClassName: pointer.String(GetStorageClassName(instanceType)),
		StorageSize:      DefaultStorageSize,
		MountPath:        MountPath,
	}
}

func buildSidecar(
	ctx context.Context,
	cfg config.RootConfig,
	cluster *cloudv1.Cluster,
	extraGroup *cloudv1.ExtraGroupCluster,
) corev1.Container {
	cmd := cfg.ClusterController.Audit.SidecarCommand
	if len(cmd) == 0 {
		cmd = []string{
			"sh",
			"-c",
			"touch /var/log/auditlog/tidb-audit.log; tail -n0 -F /var/log/auditlog/tidb-audit.log;",
		}
	}

	sidecarImage := cfg.ClusterController.Audit.SidecarImage
	sidecarEnvs := []corev1.EnvVar{
		// For the sidecar to get pod slot(ordinal) for append to the url suffix.
		// Note all pod use the same configmap as config file.
		{
			Name: "POD_NAME",
			ValueFrom: &corev1.EnvVarSource{
				FieldRef: &corev1.ObjectFieldSelector{
					FieldPath: "metadata.name",
				},
			},
		},
	}
	logger := log.FromContext(ctx)

	groupName := cluster.GetDefaultGroupID()
	if extraGroup != nil {
		groupName = extraGroup.Name
	}
	resolved := ResolveAuditSidecarImage(ctx, cluster, sidecarImage, cfg.ClusterController.Audit.BaseImage, groupName)
	sidecarImage = resolved.Image
	sidecarEnvs = append(sidecarEnvs, resolved.EnvVars...)
	logger.Info("audit sidecar image", "sidecarImage", sidecarImage, "sidecarEnvs", sidecarEnvs)

	sidecarContainer := corev1.Container{
		Name:  AuditLogName,
		Image: sdk.OverrideImageWithRepoPrefix(cfg.Override.OverrideRegistry, cfg.Override.OverrideRepoPrefix, sidecarImage),

		Command: cmd,
		// DO NOT specify request to increase the resource need by the pod.
		// Generally this sidecar should take only a pretty small resource, specify limits to avoid
		// affect the tidb pod when something unexpected take a lot of resource.
		Resources: corev1.ResourceRequirements{
			Limits: corev1.ResourceList{
				corev1.ResourceCPU:    resource.MustParse("100m"),
				corev1.ResourceMemory: resource.MustParse("256Mi"),
			},
			Requests: corev1.ResourceList{
				corev1.ResourceCPU:    resource.MustParse("0m"),
				corev1.ResourceMemory: resource.MustParse("0Mi"),
			},
		},
		VolumeMounts: []corev1.VolumeMount{
			{
				Name:      "tidb-" + SlowlogVolumeName,
				MountPath: MountPath,
			},
			{
				Name:      ConfigVolumeName,
				MountPath: ConfigMountPath(),
			},
		},
		Env: sidecarEnvs,
	}
	return sidecarContainer
}

type AuditSidecarImageResolveResult struct {
	Image   string
	EnvVars []corev1.EnvVar
}

// ResolveAuditSidecarImage resolves audit sidecar image and envs by annotation tag and optional base image.
// If the resolved sidecar version supports dynamic log path, envs are returned.
func ResolveAuditSidecarImage(
	ctx context.Context,
	cluster *cloudv1.Cluster,
	sidecarImage string,
	baseImage string,
	groupName string,
) AuditSidecarImageResolveResult {
	logger := log.FromContext(ctx)
	result := AuditSidecarImageResolveResult{
		Image: sidecarImage,
	}
	if sidecarTag := util.GetAnnotation(cluster, consts.AuditLogSidecarImageTag); sidecarTag != "" {
		if baseImage != "" {
			result.Image = OverrideSidecarImageTag(baseImage, sidecarTag)
			logger.Info("audit sidecar image overridden by base image", "baseImage", baseImage, "sidecarTag", sidecarTag, "image", result.Image)
		} else {
			result.Image = OverrideSidecarImageTag(sidecarImage, sidecarTag)
			logger.Info("audit sidecar image overridden by sidecar image", "sidecarImage", sidecarImage, "sidecarTag", sidecarTag, "image", result.Image)
		}
		if CheckIfAuditSidecarSupportDynamicLogPath(ctx, sidecarTag) {
			result.EnvVars = append(result.EnvVars, corev1.EnvVar{
				Name:  "GROUP_NAME",
				Value: groupName,
			})
			logger.Info("audit sidecar dynamic log path enabled", "groupName", groupName)
		}
	}
	return result
}

func ConfigMapName(clusterID string) string {
	return clusterID + "-audit"
}

var ConfigVolumeName = "audit-config"

func BuildConfigMapVolume(configMapName string) corev1.Volume {
	return corev1.Volume{
		Name: ConfigVolumeName,
		VolumeSource: corev1.VolumeSource{
			ConfigMap: &corev1.ConfigMapVolumeSource{
				LocalObjectReference: corev1.LocalObjectReference{
					Name: configMapName,
				},
			},
		},
	}
}

func ConfigMountPath() string {
	return filepath.Dir(configFilePath)
}

func createOrUpdateConfigMap(
	ctx context.Context,
	cluster *cloudv1.Cluster,
	tc *v1alpha1.TidbCluster,
	tcClient client.Client,
	externalID string,
) error {
	a := cluster.Spec.Cluster.TiDB.Audit
	tidbTerminationGracePeriodSeconds := consts.DefaultPodTerminationGracePeriodSeconds
	if tc.Spec.TiDB != nil && tc.Spec.TiDB.TerminationGracePeriodSeconds != nil {
		tidbTerminationGracePeriodSeconds = int(*tc.Spec.TiDB.TerminationGracePeriodSeconds)
	}
	cfg := Config{
		LogDir:                  MountPath,
		BucketURL:               a.BucketURL,
		Region:                  a.Region,
		RoleARN:                 a.RoleArn,
		ExternalID:              externalID,
		SlowLogMaxBackups:       1,
		CleanOldestLogPercent:   0.8,
		GracefulStopWaitSeconds: consts.GetAuditLogSidecarGracefulStopWaitSeconds(ctx, cluster, tidbTerminationGracePeriodSeconds),
		MetricsPort:             MetricsPort,
	}
	if a.Azure != nil && a.Azure.SASToken != "" {
		cfg.Azure = &Azure{SASToken: a.Azure.SASToken}
	}

	data, err := cfg.ToYAML()
	if err != nil {
		return err
	}

	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: tc.Namespace,
			Name:      ConfigMapName(cluster.Name),
		},
	}

	mp := map[string]string{
		"audit.yaml": string(data),
	}

	_, err = controllerutil.CreateOrUpdate(ctx, tcClient, cm, func() error {
		cm.Data = mp
		return nil
	})

	if err != nil {
		return errors.Annotate(err, "failed to create or update configmap")
	}

	return nil
}

var recheckDuration = time.Second * 5

// TODO(chenfei): remove me
// CheckIfSupported checks if the given TiDB version supports the audit log feature.
func CheckIfSupported(ctx context.Context, v string) bool {
	logger := log.FromContext(ctx)

	version, err := semver.NewVersion(v)
	if err != nil {
		logger.Error(err, "failed to parse tidb version", "version", v)
		return false
	}

	return auditLogVersionConstraint.Check(version)
}

func CheckIfAuditSidecarSupportDynamicLogPath(ctx context.Context, v string) bool {
	return CheckIfVersionSupported(ctx, auditSidecarDynamicLogPathVersionConstraint, v)
}

func CheckIfVersionSupported(ctx context.Context, constraint *semver.Constraints, v string) bool {
	logger := log.FromContext(ctx)

	version, err := semver.NewVersion(v)
	if err != nil {
		logger.Error(err, "failed to parse version for constraint", "constraint", constraint, "version", v)
		return false
	}

	return constraint.Check(version)
}

func onCreate(
	ctx context.Context,
	cfg config.RootConfig,
	cluster *cloudv1.Cluster,
	infraClient client.Client,
	tc *v1alpha1.TidbCluster,
	tcClient client.Client,
	externalID string,
	keyNs string,
	keyName string,
	extraGroup *cloudv1.ExtraGroupCluster,
) (ctrl.Result, error) {
	audit := cluster.Spec.Cluster.TiDB.Audit
	if audit == nil || !audit.Enable {
		return ctrl.Result{}, nil
	}

	if err := createOrUpdateConfigMap(ctx, cluster, tc, tcClient, externalID); err != nil {
		return ctrl.Result{}, err
	}

	if MetricsPort != 0 {
		if tc.Spec.TiDB.Annotations == nil {
			tc.Spec.TiDB.Annotations = make(map[string]string)
		}
		tc.Spec.TiDB.Annotations[MetricsPortAnnotationKey] = strconv.Itoa(MetricsPort)
	}

	storageVolume := buildAuditLogStorageVolume(audit, cluster.Spec.Cluster.TiDB.InstanceType)
	sidecar := buildSidecar(ctx, cfg, cluster, extraGroup)
	volume := BuildConfigMapVolume(ConfigMapName(cluster.Name))

	if cfg.Provider == cloudv1.ProviderTypeGKE {
		if err := setupGCPCredentials(ctx, keyNs, keyName, infraClient, tc, tcClient, &sidecar); err != nil {
			return ctrl.Result{}, err
		}
	}

	tc.Spec.TiDB.SlowLogVolumeName = storageVolume.Name
	tc.Spec.TiDB.StorageVolumes = addStorageVolumeIfNotExist(tc.Spec.TiDB.StorageVolumes, storageVolume)
	tc.Spec.TiDB.AdditionalContainers = addContainerIfNotExist(tc.Spec.TiDB.AdditionalContainers, sidecar)
	tc.Spec.TiDB.Plugins = addStringIfNotExist(tc.Spec.TiDB.Plugins, pluginName)
	tc.Spec.TiDB.AdditionalVolumes = addVolumeIfNotExist(tc.Spec.TiDB.AdditionalVolumes, volume)

	return ctrl.Result{}, nil
}

// Reconcile will create a storage class and mount a mount to tidb.
// The PVC will not be delete when cluster deleted, user need to delete it.
// tc.Spec.TiDB.SlowLogVolumeName will be set as empty if audit log is not enable.
//
// When enable will:
//  1. Create StorageClass, and configmap.
//  2. Update tc:
//     a. Add audit plugin into `spec.TiDB.Plugins`
//     b. Add sidecar into `Spec.TiDB.AdditionalContainers`
//     c. Set `Spec.TiDB.SlowLogVolumeName`
//     d. Add item into `tc.Spec.TiDB.StorageVolumes` to mount a volume using the created at step 1.
//  3. Make sure sts updated. Note `volumeClaimTemplates` of StatefulSet do not support updated, and operator will ignore it when updating STS, see: https://github.com/pingcap/tidb-operator/pull/4369. to address this:
//
// we will:
//
//	a. Delete tidb sts with DeletePropagationOrphan option to make operator recreate STS according the newest tc spec.
//	b. Delete the pods starting from slot 0 one by one to make sts recreate it. STS controller will fail to update the pod since it try to update `Volumes` field of the pod and failed.
func Reconcile(
	ctx context.Context,
	cfg config.RootConfig,
	cluster *cloudv1.Cluster,
	infraClient client.Client,
	tc *v1alpha1.TidbCluster,
	tcClient client.Client,
	advanceSTS bool,
	externalID string,
	keyNs string,
	keyName string,
	extraGroup *cloudv1.ExtraGroupCluster,
) (ctrl.Result, error) {
	if tc.CreationTimestamp.IsZero() {
		return onCreate(ctx, cfg, cluster, infraClient, tc, tcClient, externalID, keyNs, keyName, extraGroup)
	}

	logger := log.FromContext(ctx)
	oldTC := tc.DeepCopy()
	audit := cluster.Spec.Cluster.TiDB.Audit

	if err := FillSidecarContainer(ctx, cfg, cluster, infraClient, tc, tcClient, advanceSTS, externalID, keyNs, keyName, extraGroup); err != nil {
		return ctrl.Result{}, err
	}

	if !equality.Semantic.DeepEqual(oldTC.Spec, tc.Spec) {
		logger.Info("audit log related fields are changed, will update tc spec")
		err := tcClient.Update(ctx, tc)
		if err != nil {
			return ctrl.Result{}, errors.Annotate(err, "failed to update tc")
		}
	}

	if audit != nil && audit.Enable {
		stsName := tc.Name + "-tidb"
		sts := new(apps.StatefulSet)
		key := types.NamespacedName{Namespace: tc.Namespace, Name: stsName}
		var err error
		if advanceSTS {
			err = getAdvanceStatefulset(ctx, tcClient, key, sts)
		} else {
			err = tcClient.Get(ctx, key, sts)
		}
		if err != nil {
			if apierrors.IsNotFound(err) {
				return ctrl.Result{RequeueAfter: recheckDuration}, nil
			}
			return ctrl.Result{}, errors.Annotate(err, "failed to get tidb sts")
		}

		// operator will add "tidb-" prefix to volume name in tc.Spec.TiDB.StorageVolumes
		volumeName := "tidb-" + SlowlogVolumeName
		existVolume := false
		for _, entry := range sts.Spec.VolumeClaimTemplates {
			if entry.Name == volumeName {
				existVolume = true
				break
			}
		}

		if !existVolume {
			logger.Info("delete sts", "sts", sts.Name)
			var err error
			if advanceSTS {
				err = deleteAdvanceStatefuleset(ctx, tcClient, client.ObjectKeyFromObject(sts), client.PropagationPolicy(metav1.DeletePropagationOrphan))
			} else {
				err = tcClient.Delete(ctx, sts, client.PropagationPolicy(metav1.DeletePropagationOrphan))
			}
			if err != nil {
				if apierrors.IsNotFound(err) {
					// Since we get the STS first.
					// This only happen when cache client is not updated and the new STS is not recreate by operator yet.
					logger.Info("cached not updated and sts not recreate yet, keep requeue after sometime")
					return ctrl.Result{RequeueAfter: recheckDuration}, nil
				}
				return ctrl.Result{}, err
			}
			return ctrl.Result{RequeueAfter: recheckDuration}, nil
		}

		for _, ord := range sets.List(GetPodOrdinals(tc.Spec.TiDB.Replicas, sts)) {
			podName := sts.Name + "-" + strconv.Itoa(int(ord))
			po := new(corev1.Pod)
			err := tcClient.Get(ctx, types.NamespacedName{Namespace: tc.Namespace, Name: podName}, po)
			if err != nil {
				if apierrors.IsNotFound(err) {
					return ctrl.Result{RequeueAfter: recheckDuration}, nil
				}
				return ctrl.Result{}, err
			}
			isNewPod := false
			for _, v := range po.Spec.Volumes {
				if v.Name == volumeName {
					isNewPod = true
					break
				}
			}

			if !isNewPod {
				logger.Info("delete pod", "pod", po.Name)
				err := tcClient.Delete(ctx, po)
				if err != nil {
					return ctrl.Result{}, err
				}
				return ctrl.Result{RequeueAfter: recheckDuration}, nil
			}

			if !isPodReady(po) {
				logger.Info("wait pod ready", "pod", po.Name)
				return ctrl.Result{RequeueAfter: recheckDuration}, nil
			}
		}

		// all Pod is check and ready now
		if !getAuditReady(cluster) {
			setAuditReady(cluster, true)
			err = infraClient.Status().Update(ctx, cluster)
			if err != nil {
				return ctrl.Result{}, errors.Annotate(err, "failed to update cluster status")
			}
		}
		return ctrl.Result{}, nil
	}

	// Do not wait the update finished.
	if getAuditReady(cluster) {
		setAuditReady(cluster, false)
		err := infraClient.Status().Update(ctx, cluster)
		if err != nil {
			return ctrl.Result{}, errors.Annotate(err, "failed to update cluster status")
		}
	}
	return ctrl.Result{}, nil
}

func FillSidecarContainer(
	ctx context.Context,
	cfg config.RootConfig,
	cluster *cloudv1.Cluster,
	infraClient client.Client,
	tc *v1alpha1.TidbCluster,
	tcClient client.Client,
	advanceSTS bool,
	externalID string,
	keyNs string,
	keyName string,
	extraGroup *cloudv1.ExtraGroupCluster,
) error {

	audit := cluster.Spec.Cluster.TiDB.Audit
	storageVolume := buildAuditLogStorageVolume(audit, cluster.Spec.Cluster.TiDB.InstanceType)
	sidecar := buildSidecar(ctx, cfg, cluster, extraGroup)
	volume := BuildConfigMapVolume(ConfigMapName(cluster.Name))

	if audit != nil && audit.Enable {
		// Note mounted-configmaps-are-updated-automatically, the application can detect it and apply the new config.
		// https://kubernetes.io/docs/concepts/configuration/configmap/#mounted-configmaps-are-updated-automatically
		if err := createOrUpdateConfigMap(ctx, cluster, tc, tcClient, externalID); err != nil {
			return err
		}

		if cfg.Provider == cloudv1.ProviderTypeGKE {
			if err := setupGCPCredentials(ctx, keyNs, keyName, infraClient, tc, tcClient, &sidecar); err != nil {
				return err
			}
		}

		if MetricsPort != 0 {
			if tc.Spec.TiDB.Annotations == nil {
				tc.Spec.TiDB.Annotations = make(map[string]string)
			}
			tc.Spec.TiDB.Annotations[MetricsPortAnnotationKey] = strconv.Itoa(MetricsPort)
		}

		tc.Spec.TiDB.SlowLogVolumeName = storageVolume.Name
		tc.Spec.TiDB.StorageVolumes = addStorageVolumeIfNotExist(tc.Spec.TiDB.StorageVolumes, storageVolume)
		if sidecarTag := util.GetAnnotation(cluster, consts.AuditLogSidecarImageTag); sidecarTag != "" {
			tc.Spec.TiDB.AdditionalContainers = addOrUpdateContainer(tc.Spec.TiDB.AdditionalContainers, sidecar)
		} else {
			tc.Spec.TiDB.AdditionalContainers = addContainerIfNotExist(tc.Spec.TiDB.AdditionalContainers, sidecar)
		}
		tc.Spec.TiDB.Plugins = addStringIfNotExist(tc.Spec.TiDB.Plugins, pluginName)
		tc.Spec.TiDB.AdditionalVolumes = addVolumeIfNotExist(tc.Spec.TiDB.AdditionalVolumes, volume)
	} else {
		err := cleanupGCPCredentials(ctx, cluster, infraClient, tc, tcClient, &sidecar)
		if err != nil {
			return err
		}

		delete(tc.Spec.TiDB.Annotations, MetricsPortAnnotationKey)

		tc.Spec.TiDB.SlowLogVolumeName = ""
		// Do not support update, operator will just ignore it.
		// we do not force delete STS and delete PVC when disable audit log.
		tc.Spec.TiDB.StorageVolumes = deleteStorageVolume(tc.Spec.TiDB.StorageVolumes, storageVolume.Name)
		tc.Spec.TiDB.AdditionalContainers = deleteContainer(tc.Spec.TiDB.AdditionalContainers, sidecar)
		tc.Spec.TiDB.Plugins = deleteString(tc.Spec.TiDB.Plugins, pluginName)
		tc.Spec.TiDB.AdditionalVolumes = deleteVolume(tc.Spec.TiDB.AdditionalVolumes, volume.Name)
	}
	return nil
}

// OverrideSidecarImageTag replaces the tag in sidecarImage with sidecarTag.
// It ignores colons that appear before the last slash (i.e. registry host:port).
func OverrideSidecarImageTag(sidecarImage, sidecarTag string) string {
	if sidecarTag == "" {
		return sidecarImage
	}
	// Find the last colon, but ignore any colons before a slash (to avoid splitting registry host:port).
	lastColon := strings.LastIndex(sidecarImage, ":")
	lastSlash := strings.LastIndex(sidecarImage, "/")
	if lastColon > lastSlash {
		return sidecarImage[:lastColon] + ":" + sidecarTag
	}
	return sidecarImage + ":" + sidecarTag
}

func secretName(tc *v1alpha1.TidbCluster) string {
	return tc.Name + "-audit"
}

func setupGCPCredentials(
	ctx context.Context,
	ns string,
	keyName string,
	infraClient client.Client,
	tc *v1alpha1.TidbCluster,
	tcClient client.Client,
	sidecar *corev1.Container,
) error {
	data, err := GetServiceAccountKey(ctx, infraClient, ns, keyName)
	if err != nil {
		return err
	}
	sidecar.Env = append(sidecar.Env, corev1.EnvVar{
		Name:  "GOOGLE_APPLICATION_CREDENTIALS",
		Value: GCPCredentialPath,
	})

	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: tc.Namespace,
			Name:      secretName(tc),
		},
	}

	if _, err := controllerutil.CreateOrUpdate(ctx, tcClient, secret, func() error {
		secret.Data = map[string][]byte{
			filepath.Base(GCPCredentialPath): data,
		}
		return nil
	}); err != nil {
		return errors.Annotate(err, "failed to create or update secret")
	}

	v := corev1.Volume{
		Name: GCPCredentialsVolumeName,
		VolumeSource: corev1.VolumeSource{
			// ref https://github.com/kubernetes/kubernetes/issues/84880
			// DO NOT specific Items since bug of k8s 1.16 and we still using 1.16 for gcp.
			Secret: &corev1.SecretVolumeSource{
				SecretName: secretName(tc),
			},
		},
	}

	vm := corev1.VolumeMount{
		Name:      GCPCredentialsVolumeName,
		MountPath: filepath.Dir(GCPCredentialPath),
	}

	tc.Spec.TiDB.AdditionalVolumes = addVolumeIfNotExist(tc.Spec.TiDB.AdditionalVolumes, v)
	sidecar.VolumeMounts = addVolumeMountIfNotExist(sidecar.VolumeMounts, vm)

	return nil
}

func cleanupGCPCredentials(
	ctx context.Context,
	cluster *cloudv1.Cluster,
	infraClient client.Client,
	tc *v1alpha1.TidbCluster,
	tcClient client.Client,
	sidecar *corev1.Container,
) error {
	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: tc.Namespace,
			Name:      secretName(tc),
		},
	}
	err := tcClient.Delete(ctx, secret)
	if err != nil && !apierrors.IsNotFound(err) {
		return errors.Annotate(err, "failed to delete secret")
	}

	deleteVolume(tc.Spec.TiDB.AdditionalVolumes, GCPCredentialsVolumeName)

	return nil
}

func getAuditReady(cluster *cloudv1.Cluster) bool {
	if cluster.Status.DB == nil {
		return false
	}

	if cluster.Status.DB.TiDB == nil {
		return false
	}

	return cluster.Status.DB.TiDB.AuditReady
}

func setAuditReady(cluster *cloudv1.Cluster, ready bool) {
	if cluster.Status.DB == nil {
		cluster.Status.DB = &cloudv1.DBStatus{
			// The field is enum, can not be an empty string.
			SecondaryStatus: cloudv1.DBSecondaryStatusUnhealthy,
		}
	}

	if cluster.Status.DB.TiDB == nil {
		cluster.Status.DB.TiDB = new(cloudv1.TiDBStatus)
	}

	cluster.Status.DB.TiDB.AuditReady = ready
}

func isPodReady(po *corev1.Pod) bool {
	for _, c := range po.Status.Conditions {
		if c.Type == corev1.PodReady {
			return c.Status == corev1.ConditionTrue
		}
	}

	return false
}

func addStringIfNotExist(strs []string, str string) []string {
	for _, entry := range strs {
		if entry == str {
			return strs
		}
	}

	return append(strs, str)
}

func deleteString(strs []string, str string) []string {
	for idx, entry := range strs {
		if entry == str {
			return append(strs[:idx], strs[idx+1:]...)
		}
	}

	return strs
}

func addContainerIfNotExist(containers []corev1.Container, c corev1.Container) []corev1.Container {
	for _, entry := range containers {
		if entry.Name == c.Name {
			return containers
		}
	}

	return append(containers, c)
}

func addOrUpdateContainer(containers []corev1.Container, c corev1.Container) []corev1.Container {
	for idx, entry := range containers {
		if entry.Name == c.Name {
			if entry.Image != c.Image {
				containers[idx] = c
			}
			return containers
		}
	}

	return append(containers, c)
}

func deleteContainer(containers []corev1.Container, c corev1.Container) []corev1.Container {
	for idx, entry := range containers {
		if entry.Name == c.Name {
			return append(containers[:idx], containers[idx+1:]...)
		}
	}

	return containers
}

func addStorageVolumeIfNotExist(volumes []v1alpha1.StorageVolume, volume v1alpha1.StorageVolume) []v1alpha1.StorageVolume {
	for _, entry := range volumes {
		if entry.Name == volume.Name {
			return volumes
		}
	}

	return append(volumes, volume)
}

func deleteStorageVolume(volumes []v1alpha1.StorageVolume, name string) []v1alpha1.StorageVolume {
	for idx, entry := range volumes {
		if entry.Name == name {
			return append(volumes[:idx], volumes[idx+1:]...)
		}
	}

	return volumes
}

func addVolumeIfNotExist(volumes []corev1.Volume, volume corev1.Volume) []corev1.Volume {
	for _, entry := range volumes {
		if entry.Name == volume.Name {
			return volumes
		}
	}

	return append(volumes, volume)
}

func deleteVolume(volumes []corev1.Volume, name string) []corev1.Volume {
	for idx, entry := range volumes {
		if entry.Name == name {
			return append(volumes[:idx], volumes[idx+1:]...)
		}
	}

	return volumes
}

func addVolumeMountIfNotExist(vms []corev1.VolumeMount, vm corev1.VolumeMount) []corev1.VolumeMount {
	for _, entry := range vms {
		if entry.Name == vm.Name {
			return vms
		}
	}

	return append(vms, vm)
}

func GetServiceAccountKey(ctx context.Context, c client.Client, ns string, keyName string) ([]byte, error) {
	secret := corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: ns,
			Name:      keyName,
		},
	}

	if err := c.Get(ctx, client.ObjectKeyFromObject(&secret), &secret); err != nil {
		return nil, err
	}

	return secret.Data[gcpCredentialKey], nil
}