package migrationdedicated

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/avast/retry-go"
	openapiruntimeclient "github.com/go-openapi/runtime/client"
	"github.com/samber/lo"

	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/coreportalapi/model"
	_ "github.com/tidbcloud/tidb-management-service/functional-test/pkg/dataflow"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/dataflow/client/backup_service"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/dataflow/client/connector_service"
	dataflowmodel "github.com/tidbcloud/tidb-management-service/functional-test/pkg/dataflow/model"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/openapiv1beta"
	idlcluster "github.com/tidbcloud/tidb-management-service/functional-test/pkg/openapiv1beta/client/cluster"
	"github.com/tidbcloud/tidb-management-service/utilities/randstr"
)

// Test_001_XXX to Test_100_XXX are run before migration

func (su *TestDedicatedMigrationSuite) Test_001_CreateProject() {
	if su.projectID != "" {
		su.regionCidrSet = true
		su.T().Logf("projectID: %s", su.projectID)

		return
	}

	projectName := fmt.Sprintf("migration-e2e-%s", randstr.GenWithCharset(4, randstr.AlphabetCharset))
	su.projectID = su.openApiCli.CreateProject(projectName, su.cmekEnabled)
	su.Require().NotEmpty(su.projectID)
	su.T().Logf("projectName: %s, projectID: %s", projectName, su.projectID)

	if su.cmekEnabled {
		su.portalCli.CreateAwsCmek(su.orgID, su.projectID, su.region, su.customerKMSKey)
		su.T().Logf("add customerKMSKey: %s", su.customerKMSKey)
	}
}

func (su *TestDedicatedMigrationSuite) Test_002_CreateRegionCIDR() {
	if su.regionCidrSet {
		su.T().Logf("Skip creating regionCIDR, projectID: %s", su.projectID)
		return
	}
	su.portalCli.CreateCidr(su.orgID, su.projectID, cloudRegionCIDR, su.provider, su.region)
}

func (su *TestDedicatedMigrationSuite) Test_003_CreateCluster() {
	clusters := su.portalCli.ListClusters(su.orgID, su.projectID, 1, 10)
	for _, cluster := range clusters {
		switch lo.FromPtr(cluster.Name) {
		case su.clusterName:
			su.clusterID = lo.FromPtr(cluster.ID)
		case su.pausedClusterNameBeforeMig:
			su.pausedClusterIDBeforeMig = lo.FromPtr(cluster.ID)
			if *cluster.Status == model.CentralClusterStatusPAUSED {
				su.clusterBeforeMigPaused = true
			}

		case su.accesspointClusterName:
			su.accesspointClusterID = lo.FromPtr(cluster.ID)
		case su.tiflashClusterName:
			su.tiflashClusterID = lo.FromPtr(cluster.ID)
		}
	}

	tidbProfile := &model.CentralClusterComponent{
		ComponentType:   lo.ToPtr(model.CentralComponentTypeCOMPONENTTYPETIDB),
		NodeProfileName: lo.ToPtr("4C16G"),
		Replicas:        lo.ToPtr(int32(1)),
	}
	tikvProfile := &model.CentralClusterComponent{
		ComponentType:   lo.ToPtr(model.CentralComponentTypeCOMPONENTTYPETIKV),
		NodeProfileName: lo.ToPtr("4C16G"),
		Replicas:        lo.ToPtr(int32(3)),
		StorageSizeGi:   lo.ToPtr(int32(200)),
	}
	components := []*model.CentralClusterComponent{tidbProfile, tikvProfile}

	// When TiDB component has less or equal to 4 CPU cores, the TiDB Node Group is not supported
	accesspointClusterComponents := []*model.CentralClusterComponent{
		{
			ComponentType:   lo.ToPtr(model.CentralComponentTypeCOMPONENTTYPETIDB),
			NodeProfileName: lo.ToPtr("8C16G"),
			Replicas:        lo.ToPtr(int32(1)),
		},
		{
			ComponentType:   lo.ToPtr(model.CentralComponentTypeCOMPONENTTYPETIKV),
			NodeProfileName: lo.ToPtr("8C32G"),
			Replicas:        lo.ToPtr(int32(3)),
			StorageSizeGi:   lo.ToPtr(int32(200)),
		},
	}

	tiflashClusterComponents := []*model.CentralClusterComponent{
		{
			ComponentType:   lo.ToPtr(model.CentralComponentTypeCOMPONENTTYPETIDB),
			NodeProfileName: lo.ToPtr("8C16G"),
			Replicas:        lo.ToPtr(int32(1)),
		},
		{
			ComponentType:   lo.ToPtr(model.CentralComponentTypeCOMPONENTTYPETIKV),
			NodeProfileName: lo.ToPtr("8C32G"),
			Replicas:        lo.ToPtr(int32(3)),
			StorageSizeGi:   lo.ToPtr(int32(200)),
		},
		{
			ComponentType:   lo.ToPtr(model.CentralComponentTypeCOMPONENTTYPETIFLASH),
			NodeProfileName: lo.ToPtr("8C64G"),
			Replicas:        lo.ToPtr(int32(1)),
			StorageSizeGi:   lo.ToPtr(int32(200)),
		},
	}

	if su.addTiflash {
		if su.tiflashClusterID == "" {
			su.tiflashClusterID = *su.portalCli.CreateCluster(su.orgID, su.projectID, su.provider, su.region, su.tiflashClusterName, 4000, tiflashClusterComponents, false, 0)
			su.Require().NotEmpty(su.tiflashClusterID)
			su.T().Logf("tiflashClusterID: %s", su.tiflashClusterID)
		} else {
			su.T().Logf("Skip creating tiflash cluster, tiflashClusterID: %s", su.tiflashClusterID)
		}
	}

	if su.clusterID == "" {
		su.clusterID = *su.portalCli.CreateCluster(su.orgID, su.projectID, su.provider, su.region, su.clusterName, 4000, components, false, 0)
		su.Require().NotEmpty(su.clusterID)
		su.T().Logf("clusterID: %s", su.clusterID)
	} else {
		su.T().Logf("Skip creating cluster, clusterID: %s", su.clusterID)
	}

	if su.pausedClusterIDBeforeMig == "" && !su.skipPauseClusterCase {
		su.pausedClusterIDBeforeMig = *su.portalCli.CreateCluster(su.orgID, su.projectID, su.provider, su.region, su.pausedClusterNameBeforeMig, 4000, components, false, 0)
		su.Require().NotEmpty(su.pausedClusterIDBeforeMig)
		su.T().Logf("pausedClusterIDBeforeMig: %s", su.pausedClusterIDBeforeMig)
	} else {
		su.T().Logf("Skip creating paused cluster, pausedClusterIDBeforeMig: %s", su.pausedClusterIDBeforeMig)
	}

	if su.provider == "aws" && !su.cmekEnabled && su.accesspointClusterID == "" && !su.skipAccesspoint {
		su.accesspointClusterID = *su.portalCli.CreateCluster(su.orgID, su.projectID, su.provider, su.region, su.accesspointClusterName, 4000, accesspointClusterComponents, false, 0)
		su.Require().NotEmpty(su.accesspointClusterID)
		su.T().Logf("accesspointClusterID: %s", su.accesspointClusterID)
	} else {
		su.T().Logf("Skip creating accesspoint cluster, accesspointClusterID: %s", su.accesspointClusterID)
	}
}

func (su *TestDedicatedMigrationSuite) Test_004_PollClusterStatus() {
	su.waitForClusterStatusNormal(model.CentralClusterStatusNORMAL, su.clusterID, su.clusterName)

	if su.pausedClusterIDBeforeMig != "" && !su.clusterBeforeMigPaused {
		su.waitForClusterStatusNormal(
			model.CentralClusterStatusNORMAL,
			su.pausedClusterIDBeforeMig,
			su.pausedClusterNameBeforeMig,
		)
	}

	if su.accesspointClusterID != "" {
		su.waitForClusterStatusNormal(
			model.CentralClusterStatusNORMAL,
			su.accesspointClusterID,
			su.accesspointClusterName,
		)
	}

	resp := su.portalCli.GetCluster(su.orgID, su.projectID, su.clusterID)
	su.publicEndpointHost = lo.FromPtr(resp.Endpoint.Address)
	su.vpcPeeringEndpointHost = lo.FromPtr(resp.PrivateEndpoint.Address)
	su.Require().NotEmpty(su.publicEndpointHost)
	su.Require().NotEmpty(su.vpcPeeringEndpointHost)

	su.portalCli.SetPassword(su.orgID, su.projectID, su.clusterID, su.password)
	retryErr := retry.Do(
		func() error {
			response, err := su.portalCli.SetPublicInternetAccess(su.orgID, su.projectID, su.clusterID)
			if err != nil {
				return err
			}
			if response.Payload == nil {
				return errors.New("response payload is nil")
			}
			if response.Payload.BaseResp.ErrCode != "0" {
				return fmt.Errorf("response error code: %s", response.Payload.BaseResp.ErrCode)
			}
			return nil
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
}

func (su *TestDedicatedMigrationSuite) Test_005_PauseCluster() {
	if su.skipPauseClusterCase {
		su.T().Logf("Skip pausing cluster, clusterID: %s", su.pausedClusterIDBeforeMig)
		return
	}

	if su.clusterBeforeMigPaused {
		su.T().Logf("Skip pausing cluster, clusterID: %s is already paused", su.pausedClusterIDBeforeMig)
		return
	}

	su.portalCli.PauseCluster(su.orgID, su.projectID, su.pausedClusterIDBeforeMig)
}

func (su *TestDedicatedMigrationSuite) Test_006_CreateAccesspoint() {
	if su.accesspointClusterID == "" {
		return
	}

	listAccessPointResp := su.openApiCli.ListAccessPoint(su.projectID, su.accesspointClusterID)
	for _, item := range listAccessPointResp.Items {
		if lo.FromPtr(item.DisplayName) == "e2e-test" {
			su.accesspointID = lo.FromPtr(item.ID)
			break
		}
	}

	if su.accesspointID != "" {
		su.T().Logf("Skip creating accesspoint, accesspointID: %s", su.accesspointID)
		return
	}

	// create accesspoint
	accesspointResp, err := su.openApiCli.CreateAccessPoint(su.projectID, su.accesspointClusterID)
	su.Require().NoError(err)
	su.Require().NotNil(accesspointResp)
	su.accesspointID = accesspointResp.ID
	su.T().Logf("accesspointID: %s", su.accesspointID)

	// scale accesspoint
	resp, err := su.openApiCli.ScaleClusterByAccessPoint(
		su.projectID,
		su.accesspointClusterID,
		su.accesspointID,
		&openapiv1beta.ScaleTiDBConfig{
			NodeQuantity: 1,
		},
	)
	su.Require().True(resp.IsSuccess())
	su.Require().NoError(err)
}

func (su *TestDedicatedMigrationSuite) Test_007_CreatePrivateLink() {
	var privateLinkService *model.CentralPrivateLinkService

	// create private link service for gcp cluster
	if su.provider == "gcp" {
		privateLinkService = su.portalCli.GetPrivateLinkService(su.orgID, su.projectID, su.clusterID)
		su.Require().NotEmpty(privateLinkService)

		if lo.FromPtr(privateLinkService.Status) == model.CentralPrivateLinkServiceStatusNOTCONFIGURE {
			su.portalCli.CreatePrivateLinkService(su.orgID, su.projectID, su.clusterID)
		}
	}

	// check status of private link service
	for range 15 {
		privateLinkService = su.portalCli.GetPrivateLinkService(su.orgID, su.projectID, su.clusterID)
		su.Require().NotEmpty(privateLinkService)

		if lo.FromPtr(privateLinkService.Status) == model.CentralPrivateLinkServiceStatusACTIVE {
			break
		}
		su.T().Logf("Private Link Service status: %s", lo.FromPtr(privateLinkService.Status))
		time.Sleep(time.Second * 60)
	}
	su.Require().NotNil(privateLinkService)
	serviceName := privateLinkService.Name
	su.Require().NotEmpty(serviceName)
	dnsName := privateLinkService.DNSName
	su.Require().NotEmpty(dnsName)

	// set private link endpoint host
	switch su.provider {
	case "aws":
		su.privateLinkEndpointHost = dnsName
	case "gcp":
		su.privateLinkEndpointHost = fmt.Sprintf("%s.%s", su.gcpPrivateLinkEndpointName, dnsName)
	}

	resp := su.openApiCli.ListClusterPrivateLinkEndpoints(su.projectID, su.clusterID)
	for _, item := range resp.Endpoints {
		su.privateLinkEndpointID = item.ID
		switch su.provider {
		case "aws":
			su.awsVpcEndpointName = item.EndpointName
		case "gcp":
			su.gcpVpcEndpointName = item.EndpointName
			re := regexp.MustCompile(`^projects/.+/regions/.+/forwardingRules/(.+)$`)
			matches := re.FindStringSubmatch(su.gcpVpcEndpointName)
			if len(matches) > 1 {
				su.gcpPrivateLinkEndpointName = matches[1]
			}
		}
	}

	if su.privateLinkEndpointID != "" {
		su.T().Logf("Skip creating private link endpoint, privateLinkEndpointID: %s", su.privateLinkEndpointID)
		return
	}

	// create vpc endpoint
	var endpointName string
	switch su.provider {
	case "aws":
		endpointName = su.createAwsPrivateLink(serviceName)
		su.T().Logf("AWS private link vpcEndpointID: %s", endpointName)
		su.awsVpcEndpointName = endpointName
	case "gcp":
		endpointName = su.createGcpPrivateLink(serviceName)
		su.gcpVpcEndpointName = endpointName
	}
	su.Require().NotEmpty(endpointName)

	// create private link endpoint
	privateLinkEndpoint := su.portalCli.CreatePrivateLinkEndpoint(su.orgID, su.projectID, su.clusterID, endpointName)
	su.Require().NotEmpty(privateLinkEndpoint)

	// check status of private link endpoint
	var endpoint *idlcluster.ListPrivateEndpointsOKBodyEndpointsItems0
	retryErr := retry.Do(
		func() error {
			listPrivateLinkEndpointsResp := su.openApiCli.ListClusterPrivateLinkEndpoints(su.projectID, su.clusterID)
			var ok bool
			endpoint, ok = lo.Find(
				listPrivateLinkEndpointsResp.Endpoints,
				func(endpoint *idlcluster.ListPrivateEndpointsOKBodyEndpointsItems0) bool {
					return endpoint != nil && endpoint.EndpointName == endpointName
				},
			)
			if ok && endpoint.Status == "ACTIVE" {
				return nil
			}
			if !ok {
				su.T().Logf("Private link endpoint not found: %s", endpointName)
				return fmt.Errorf("private link endpoint not found: %s", endpointName)
			}
			su.T().Logf("Private link endpoint status: %s", endpoint.Status)
			return fmt.Errorf("private link endpoint status: %s", endpoint.Status)
		},
		retry.Attempts(90),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(10*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotNil(endpoint)
	su.privateLinkEndpointID = endpoint.ID

	su.Require().NotEmpty(su.privateLinkEndpointID)
	su.T().Logf("privateLinkEndpointID: %s", su.privateLinkEndpointID)

	// modify vpc endpoint to enable private dns
	if su.provider == "aws" {
		su.createAwsPrivateDns(endpointName)
	}
}

func (su *TestDedicatedMigrationSuite) createGcpPrivateLink(serviceName string) string {
	createVpcEndpointCmd := fmt.Sprintf(
		"gcloud compute addresses create %s --region=%s --subnet=%s --project=%s\n\n"+
			"gcloud compute forwarding-rules create %s --region=%s  --network=%s  --address=%s --target-service-attachment=%s  --service-directory-registration=projects/%s/locations/%s/namespaces/tidb%s --project=%s",
		su.gcpPrivateLinkEndpointName,
		su.region,
		gcpPrivateLinkSubnetName,
		gcpCustomerProjectID,
		su.gcpPrivateLinkEndpointName,
		su.region,
		gcpPrivateLinkNetworkName,
		su.gcpPrivateLinkEndpointName,
		serviceName,
		gcpCustomerProjectID,
		su.region,
		su.clusterID,
		gcpCustomerProjectID,
	)
	su.T().Logf("createVpcEndpointCmd: %s", createVpcEndpointCmd)
	var (
		createVpcEndpointCmdOutput string
		err                        error
	)
	retryErr := retry.Do(
		func() error {
			createVpcEndpointCmdOutput, err = su.runCommandOnGcpPrivateLinkVm(createVpcEndpointCmd)
			su.T().Logf("createVpcEndpointCmdOutput: %s", createVpcEndpointCmdOutput)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(createVpcEndpointCmdOutput)

	var endpointName string
	if retryErr == nil {
		endpointName = fmt.Sprintf(
			"projects/%s/regions/%s/forwardingRules/%s",
			gcpCustomerProjectID,
			su.region,
			su.gcpPrivateLinkEndpointName,
		)
	}

	return endpointName
}

func (su *TestDedicatedMigrationSuite) createAwsPrivateLink(serviceName string) string {
	createVpcEndpointCmd := fmt.Sprintf(
		"aws ec2 create-vpc-endpoint --vpc-id %s --region %s --service-name %s --vpc-endpoint-type Interface --subnet-ids %s --query 'VpcEndpoint.VpcEndpointId'",
		awsCustomerVpcID,
		su.region,
		serviceName,
		awsCustomerVpcSubnetIds,
	)
	su.T().Logf("createVpcEndpointCmd: %s", createVpcEndpointCmd)
	var (
		createVpcEndpointCmdOutput string
		err                        error
	)
	retryErr := retry.Do(
		func() error {
			createVpcEndpointCmdOutput, err = su.runCommandOnAwsEc2(createVpcEndpointCmd)
			su.T().Logf("createVpcEndpointCmdOutput: %s", createVpcEndpointCmdOutput)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(createVpcEndpointCmdOutput)
	stringSlice := strings.Split(createVpcEndpointCmdOutput, "\"")
	var vpcEndpointID string
	if len(stringSlice) > 1 {
		vpcEndpointID = stringSlice[1]
	}
	su.Require().NotEmpty(vpcEndpointID)

	return vpcEndpointID
}

func (su *TestDedicatedMigrationSuite) createAwsPrivateDns(endpointName string) {
	modifyVpcEndpointCmd := fmt.Sprintf(
		"aws ec2 modify-vpc-endpoint --vpc-endpoint-id %s --region %s --private-dns-enabled",
		endpointName,
		su.region,
	)
	su.T().Logf("modifyVpcEndpointCmd: %s", modifyVpcEndpointCmd)
	var (
		modifyVpcEndpointCmdOutput string
		err                        error
	)
	retryErr := retry.Do(
		func() error {
			modifyVpcEndpointCmdOutput, err = su.runCommandOnAwsEc2(modifyVpcEndpointCmd)
			su.T().Logf("modifyVpcEndpointCmdOutput: %s", modifyVpcEndpointCmdOutput)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(modifyVpcEndpointCmdOutput)
}

func (su *TestDedicatedMigrationSuite) Test_008_CreateVpcPeering() {
	switch su.provider {
	case "aws":
		vpcPeerings := su.portalCli.ListAwsVpcPeerings(su.orgID, su.projectID)
		for _, vpcPeering := range vpcPeerings.Items {
			su.dbaasAwsVpcPeeringID = lo.FromPtr(vpcPeering.ID)
			su.Require().Equal(su.region, lo.FromPtr(vpcPeering.Region))
			su.Require().Equal("active", lo.FromPtr(vpcPeering.Status))
			break
		}

		if su.dbaasAwsVpcPeeringID != "" {
			su.T().Logf("Skip creating aws vpc peering, dbaasAwsVpcPeeringID: %s", su.dbaasAwsVpcPeeringID)
			return
		}
		su.createAwsVpcPeering()
	case "gcp":
		vpcPeerings := su.portalCli.ListGcpVpcPeerings(su.orgID, su.projectID)
		for _, vpcPeering := range vpcPeerings.Items {
			su.dbaasGcpVpcPeeringID = lo.FromPtr(vpcPeering.ID)
			su.Require().Equal(su.region, lo.FromPtr(vpcPeering.Region))
			su.Require().Equal("active", lo.FromPtr(vpcPeering.Status))
			break
		}

		if su.dbaasGcpVpcPeeringID != "" {
			su.T().Logf("Skip creating gcp vpc peering, dbaasGcpVpcPeeringID: %s", su.dbaasGcpVpcPeeringID)
			return
		}
		su.createGcpVpcPeering()
	}
}

func (su *TestDedicatedMigrationSuite) createAwsVpcPeering() {
	// create vpc peering
	createAWSVpcPeeringResp := su.portalCli.CreateAWSVpcPeering(
		su.orgID,
		su.projectID,
		su.region,
		su.region,
		awsCumstomerAccountID,
		awsCumstomerVpcCidr,
		awsCustomerVpcID,
	)
	su.Require().NotEmpty(createAWSVpcPeeringResp)

	vpcPeeringID := lo.FromPtr(createAWSVpcPeeringResp.ID)
	su.Require().NotEmpty(vpcPeeringID)
	su.dbaasAwsVpcPeeringID = vpcPeeringID
	su.T().Logf("aws vpcPeeringID on dbaas: %s", vpcPeeringID)

	// query aws vpc peering id
	var awsVpcPeeringID string
	retryErr := retry.Do(
		func() error {
			vpcPeering := su.portalCli.GetAwsVpcPeering(su.orgID, su.projectID, vpcPeeringID)
			if lo.FromPtr(vpcPeering.VpcPeeringID) != "" {
				awsVpcPeeringID = lo.FromPtr(vpcPeering.VpcPeeringID)
				return nil
			}
			return errors.New("aws vpc peering id is empty")
		},
		retry.Attempts(50),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(30*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(awsVpcPeeringID)
	su.T().Logf("awsVpcPeeringID: %s", awsVpcPeeringID)

	// accept vpc peering
	acceptVpcPeeringCmd := fmt.Sprintf(
		"aws ec2 accept-vpc-peering-connection --vpc-peering-connection-id %s --region %s",
		awsVpcPeeringID,
		su.region,
	)
	su.T().Logf("acceptVpcPeeringCmd: %s", acceptVpcPeeringCmd)
	var (
		acceptVpcPeeringCmdOutput string
	)
	retryErr = retry.Do(
		func() error {
			var err error
			acceptVpcPeeringCmdOutput, err = su.runCommandOnAwsEc2(acceptVpcPeeringCmd)
			su.T().Logf("acceptVpcPeeringCmdOutput: %s", acceptVpcPeeringCmdOutput)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(acceptVpcPeeringCmdOutput)

	// check vpc peering status
	retryErr = retry.Do(
		func() error {
			vpcPeering := su.portalCli.GetAwsVpcPeering(su.orgID, su.projectID, vpcPeeringID)
			su.T().Logf("VPC peering status: %s", lo.FromPtr(vpcPeering.Status))
			if lo.FromPtr(vpcPeering.Status) == "active" {
				return nil
			}
			return fmt.Errorf("VPC peering status: %s", lo.FromPtr(vpcPeering.Status))
		},
		retry.Attempts(50),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(30*time.Second),
	)
	su.Require().NoError(retryErr)

	// delete vpc peering route, ignore error
	su.deleteAwsVpcPeeringRoute()

	// add vpc peering route
	su.addAwsVpcPeeringRoute(awsVpcPeeringID)
}

func (su *TestDedicatedMigrationSuite) deleteAwsVpcPeeringRoute() {
	cmd := fmt.Sprintf(
		"aws ec2 describe-route-tables --region %s --filters Name=vpc-id,Values=\"%q\" --query 'RouteTables[*].RouteTableId' --output text | tr \"\\t\" \"\\n\" | while read row\ndo\n    app_route_table_id=\"$row\"\n    aws ec2 delete-route --route-table-id \"$app_route_table_id\" --destination-cidr-block \"%q\" --region %s\ndone",
		su.region,
		awsCustomerVpcID,
		cloudRegionCIDR,
		su.region,
	)
	su.T().Logf("DeleteAwsVpcPeeringRouteCmd: %s", cmd)
	var (
		output string
	)
	_ = retry.Do(
		func() error {
			var err error
			output, err = su.runCommandOnAwsEc2(cmd)
			su.T().Logf("DeleteAwsVpcPeeringRouteCmdOutput: %s", output)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
}

func (su *TestDedicatedMigrationSuite) addAwsVpcPeeringRoute(awsVpcPeeringID string) {
	cmd := fmt.Sprintf(
		"aws ec2 describe-route-tables --region %s --filters Name=vpc-id,Values=\"%q\" --query 'RouteTables[*].RouteTableId' --output text | tr \"\\t\" \"\\n\" | while read row\ndo\n    app_route_table_id=\"$row\"\n    aws ec2 create-route --route-table-id \"$app_route_table_id\" --destination-cidr-block \"%q\" --vpc-peering-connection-id \"%q\" --region %s\ndone",
		su.region,
		awsCustomerVpcID,
		cloudRegionCIDR,
		awsVpcPeeringID,
		su.region,
	)
	su.T().Logf("AddAwsVpcPeeringRouteCmd: %s", cmd)
	var (
		output string
	)
	retryErr := retry.Do(
		func() error {
			var err error
			output, err = su.runCommandOnAwsEc2(cmd)
			su.T().Logf("AddAwsVpcPeeringRouteCmdOutput: %s", output)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
}

func (su *TestDedicatedMigrationSuite) createGcpVpcPeering() {
	// delete old vpc peering and ignore error
	deleteVpcPeeringCmd := fmt.Sprintf(
		"gcloud compute networks peerings delete %s --network=dedicated-migration-vpc-peering",
		gcpVpcPeeringName,
	)
	su.T().Logf("deleteVpcPeeringCmd: %s", deleteVpcPeeringCmd)
	var (
		deleteVpcPeeringCmdOutput string
	)
	_ = retry.Do(
		func() error {
			var err error
			deleteVpcPeeringCmdOutput, err = su.runCommandOnGcpVpcPeeringVm(deleteVpcPeeringCmd)
			su.T().Logf("deleteVpcPeeringCmdOutput: %s, err: %v", deleteVpcPeeringCmdOutput, err)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)

	// create vpc peering
	createGcpVpcPeeringResp := su.portalCli.CreateGcpVpcPeering(
		su.orgID,
		su.projectID,
		su.region,
		su.region,
		gcpCustomerProjectID,
		gcpCustomerVpcCidr,
		gcpVpcPeeringNetworkName,
	)

	su.Require().NotEmpty(createGcpVpcPeeringResp)
	vpcPeeringID := lo.FromPtr(createGcpVpcPeeringResp.ID)
	su.Require().NotEmpty(vpcPeeringID)

	su.dbaasGcpVpcPeeringID = vpcPeeringID
	su.T().Logf("GCP vpcPeeringID on dbaas: %s", vpcPeeringID)

	var (
		peerProject string
		peerNetwork string
	)
	retryErr := retry.Do(
		func() error {
			vpcPeering := su.portalCli.GetGcpVpcPeering(su.orgID, su.projectID, vpcPeeringID)
			if lo.FromPtr(vpcPeering.ProjectID) != "" && lo.FromPtr(vpcPeering.NetworkName) != "" {
				peerProject = lo.FromPtr(vpcPeering.ProjectID)
				peerNetwork = lo.FromPtr(vpcPeering.NetworkName)
				return nil
			}
			return errors.New("peer project or network is empty")
		},
		retry.Attempts(12),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(peerProject)
	su.Require().NotEmpty(peerNetwork)
	su.T().Logf("peerProject: %s, peerNetwork: %s", peerProject, peerNetwork)

	// create vpc peering on gcp
	createVpcPeeringCmd := fmt.Sprintf(
		"gcloud beta compute networks peerings create %s --project %s --network %s --peer-project %s --peer-network %s",
		gcpVpcPeeringName,
		gcpCustomerProjectID,
		gcpVpcPeeringNetworkName,
		peerProject,
		peerNetwork,
	)
	su.T().Logf("createVpcPeeringCmd: %s", createVpcPeeringCmd)
	var (
		createVpcPeeringCmdOutput string
	)
	retryErr = retry.Do(
		func() error {
			var err error
			createVpcPeeringCmdOutput, err = su.runCommandOnGcpVpcPeeringVm(createVpcPeeringCmd)
			su.T().Logf("createVpcPeeringCmdOutput: %s", createVpcPeeringCmdOutput)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(createVpcPeeringCmdOutput)

	// check vpc peering status
	retryErr = retry.Do(
		func() error {
			vpcPeering := su.portalCli.GetGcpVpcPeering(su.orgID, su.projectID, vpcPeeringID)
			if lo.FromPtr(vpcPeering.Status) == "active" {
				return nil
			}
			return fmt.Errorf("vpc peering status: %s", lo.FromPtr(vpcPeering.Status))
		},
		retry.Attempts(60),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(10*time.Second),
	)
	su.Require().NoError(retryErr)
}

func (su *TestDedicatedMigrationSuite) getGCPAccountID(orgID, projectID, clusterID string) (string, error) {
	params := connector_service.NewConnectorServiceDescribeAdminProfileParams().WithDefaults()
	params.SetClusterID(clusterID)
	params.SetProjectID(projectID)
	params.SetOrgID(orgID)
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	resp, err := su.dataflowCli.ConnectorService.ConnectorServiceDescribeAdminProfile(params, authInfoWriter)
	if err != nil {
		return "", err
	}
	gcpAccountID := lo.FromPtr(resp.GetPayload().GcpAccountID)
	if gcpAccountID == "" {
		return "", fmt.Errorf("gcp_account_id for cluster %s is empty", clusterID)
	}
	return gcpAccountID, nil
}

func (su *TestDedicatedMigrationSuite) allowSeviceAccountAccessGCSBucket(serviceAccount string) error {
	addIAMPolicyBindingCmd := fmt.Sprintf(
		"gcloud storage buckets add-iam-policy-binding gs://dedicated-migration-cdc-us-west1 --member=serviceAccount:%s --role=projects/pingcap-gardener/roles/CustomRole352",
		serviceAccount,
	)
	su.T().Logf("addIAMPolicyBindingCmd: %s", addIAMPolicyBindingCmd)
	var cmdOutput string
	retryErr := retry.Do(
		func() error {
			var err error
			cmdOutput, err = su.runCommandOnGcpPrivateLinkVm(addIAMPolicyBindingCmd)
			su.T().Logf("addIAMPolicyBindingCmdOutput: %s", cmdOutput)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	if cmdOutput == "" {
		return errors.New("addIAMPolicyBindingCmdOutput is empty")
	}
	return retryErr
}

func (su *TestDedicatedMigrationSuite) Test_009_CreateTiCDCIfNotExisted() {
	name := "mig-e2e-changefeed"
	// list before create
	{
		params := connector_service.NewListConnectorsParams().WithDefaults()
		params.SetClusterID(su.clusterID)
		params.SetProjectID(su.projectID)
		params.SetOrgID(su.orgID)
		params.SetPage(1)
		params.SetPerPage(10)
		authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
		resp, err := su.dataflowCli.ConnectorService.ListConnectors(params, authInfoWriter)
		su.Require().NoError(err)
		// su.T().Logf("list connectors, params:%#v, len(items):%v", params, len(resp.GetPayload().Items))
		var existedConnectorID string
		for _, item := range resp.GetPayload().Items {
			// su.T().Logf("connector, name:%s, status:%v, id:%v", item.Name, *item.Status, *item.ConnectorID)
			if item.Name == name && lo.FromPtr(item.Status) == dataflowmodel.DataflowConnectorStatusRUNNING {
				existedConnectorID = *item.ConnectorID
				break
			}
		}
		if existedConnectorID != "" {
			su.connectIDBeforeMig = existedConnectorID
			su.T().Logf("changefeed %s is already existed, skip the creation", name)
			return
		}
	}
	params := connector_service.NewCreateConnectorParams()
	params.SetClusterID(su.clusterID)
	params.SetProjectID(su.projectID)
	params.SetOrgID(su.orgID)
	body := connector_service.CreateConnectorBody{}
	switch su.provider {
	case "aws":
		s3URI := fmt.Sprintf("s3://dedicated-migration-cdc-us-west-2/%s/", su.clusterID)
		body.ConnectorType = dataflowmodel.DataflowConnectorTypeS3.Pointer()
		body.ChangefeedInstanceType = "4rcu"
		body.URL = lo.ToPtr(s3URI)
		body.Name = name
		body.S3Info = new(dataflowmodel.DataflowS3Info)
		body.S3Info.Endpoint = &dataflowmodel.DataflowS3Endpoint{
			S3URI:     s3URI,
			AccessKey: su.s3AK,
			SecretKey: su.s3SK,
		}
		body.S3Info.DataFormat = &dataflowmodel.DataflowS3Dataformat{
			Protocol:             dataflowmodel.DataflowS3ProtocolS3PROTOCOLCSV.Pointer(),
			Delimiter:            ",",
			BinaryEncodingMethod: dataflowmodel.DataflowBinaryEncodingMethodBASE64.Pointer(),
			Quote:                "\"",
			Null:                 "\\N",
		}
		body.S3Info.StartPosition = &dataflowmodel.DataflowStartPosition{
			Mode: dataflowmodel.DataflowStartModeFROMNOW.Pointer(),
		}
		body.S3Info.StorageFlushConfig = &dataflowmodel.DataflowStorageFlushConfig{
			FlushInterval: "60s",
			FileSize:      "67108864",
		}
		body.S3Info.TableConfig = &dataflowmodel.DataflowTableConfig{
			Mode:       dataflowmodel.DataflowTableModeFORCESYNC.Pointer(),
			FilterRule: []string{"*.*"},
		}
	case "gcp":
		// get admin profile first
		// the gcp account id is service account
		serviceAccount, err := su.getGCPAccountID(su.orgID, su.projectID, su.clusterID)
		su.Require().NoError(err)
		err = su.allowSeviceAccountAccessGCSBucket(serviceAccount)
		su.Require().NoError(err)

		gcsURI := fmt.Sprintf("gs://dedicated-migration-cdc-us-west1/%s/", su.clusterID)
		body.ConnectorType = dataflowmodel.DataflowConnectorTypeGCS.Pointer()
		body.ChangefeedInstanceType = "2rcu"
		body.URL = lo.ToPtr(gcsURI)
		body.Name = name
		body.GcsInfo = new(dataflowmodel.DataflowGcsInfo)
		body.GcsInfo.Endpoint = &dataflowmodel.DataflowGcsEndpoint{
			GcsURI: gcsURI,
		}
		body.GcsInfo.DataFormat = &dataflowmodel.DataflowS3Dataformat{
			Protocol:             dataflowmodel.DataflowS3ProtocolS3PROTOCOLCSV.Pointer(),
			Delimiter:            ",",
			BinaryEncodingMethod: dataflowmodel.DataflowBinaryEncodingMethodBASE64.Pointer(),
			Quote:                "\"",
			Null:                 "\\N",
		}
		body.GcsInfo.StartPosition = &dataflowmodel.DataflowStartPosition{
			Mode: dataflowmodel.DataflowStartModeFROMNOW.Pointer(),
		}
		body.GcsInfo.StorageFlushConfig = &dataflowmodel.DataflowStorageFlushConfig{
			FlushInterval: "60s",
			FileSize:      "67108864",
		}
		body.GcsInfo.TableConfig = &dataflowmodel.DataflowTableConfig{
			Mode:       dataflowmodel.DataflowTableModeFORCESYNC.Pointer(),
			FilterRule: []string{"*.*"},
		}
	}
	params.SetBody(body)
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	var resp *connector_service.CreateConnectorOK
	retryErr := retry.Do(
		func() error {
			var err error
			resp, err = su.dataflowCli.ConnectorService.CreateConnector(params, authInfoWriter)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotNil(resp.GetPayload().ConnectorID)
	su.connectIDBeforeMig = *resp.GetPayload().ConnectorID
	su.T().Logf("changefeed %s is created, wait it to be ready", name)
	{
		params := connector_service.NewGetConnectorParams()
		params.SetClusterID(su.clusterID)
		params.SetProjectID(su.projectID)
		params.SetOrgID(su.orgID)
		params.SetConnectorID(su.connectIDBeforeMig)
		authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
		for {
			time.Sleep(1 * time.Minute)
			resp, err := su.dataflowCli.ConnectorService.GetConnector(params, authInfoWriter)
			su.Require().NoError(err)
			status := *(resp.GetPayload().Status)
			su.T().Logf("changefeed %s status is %s", name, status)
			if status == dataflowmodel.DataflowConnectorStatusRUNNING {
				break
			}
		}
	}
}

func (su *TestDedicatedMigrationSuite) Test_010_CreateBackup() {
	backupName := "bak-before-mig"

	{
		params := backup_service.NewListBackupsParams()
		params.SetClusterID(su.clusterID)
		params.SetProjectID(su.projectID)
		params.SetOrgID(su.orgID)
		params.SetPage(1)
		params.SetPerPage(10)
		authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
		resp, err := su.dataflowCli.BackupService.ListBackups(params, authInfoWriter)
		su.Require().NoError(err, "ListBackups failed")

		for _, item := range resp.Payload.Items {
			if lo.FromPtr(item.Name) == backupName &&
				lo.FromPtr(item.Status) == dataflowmodel.BackupExternalStatusSUCCESS {
				su.bakIDBeforeMig = lo.FromPtr(item.ID)
				break
			}
		}
	}

	if su.bakIDBeforeMig != "" {
		su.T().Logf("Skip creating backup, bakIDBeforeMig: %s", su.bakIDBeforeMig)
		return
	}

	currUTC := time.Now().UTC()
	backupTime := currUTC.Format(time.RFC3339)

	retryErr := retry.Do(
		func() error {
			response, err := su.dataflowCli.CreateBackupSchedule(
				su.orgID,
				su.projectID,
				su.clusterID,
				backupName,
				backupTime,
			)
			if err != nil {
				return err
			}
			if response.Payload == nil {
				return errors.New("response payload is nil")
			}
			su.bakIDBeforeMig = response.GetPayload().BackupID
			return nil
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(su.bakIDBeforeMig)
	su.T().Logf("bakIDBeforeMig is %s", su.bakIDBeforeMig)

	{
		params := backup_service.NewListBackupsParams()
		params.SetClusterID(su.clusterID)
		params.SetProjectID(su.projectID)
		params.SetOrgID(su.orgID)
		params.SetBackupID(&su.bakIDBeforeMig)
		params.SetPage(1)
		params.SetPerPage(10)
		authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
		for {
			time.Sleep(30 * time.Second)
			resp, err := su.dataflowCli.BackupService.ListBackups(params, authInfoWriter)
			su.Require().NoError(err, "ListBackups failed")
			su.Require().NotEmpty(resp.Payload.Items)
			bak := resp.Payload.Items[0]
			su.Require().NotEqual(dataflowmodel.BackupExternalStatusFAILED, *bak.Status)
			su.T().Logf("backup %s status: %s", su.bakIDBeforeMig, *bak.Status)
			if *bak.Status == dataflowmodel.BackupExternalStatusSUCCESS {
				break
			}
		}
	}
}

func (su *TestDedicatedMigrationSuite) Test_011_WaitForAccesspointReady() {
	if su.accesspointClusterID == "" {
		return
	}

	su.waitForClusterStatusNormal(model.CentralClusterStatusNORMAL, su.accesspointClusterID, su.accesspointClusterName)

	// set password
	su.portalCli.SetPassword(su.orgID, su.projectID, su.accesspointClusterID, su.accesspointClusterPassword)
	su.T().Logf("password was set for cluster %s", su.accesspointClusterID)

	// check status of private link service
	var privateLinkService *idlcluster.GetAccesspointOKBodyPrivateEndpointPrivateLinkService
	for range 15 {
		resp := su.openApiCli.GetAccessPoint(su.projectID, su.accesspointClusterID, su.accesspointID)
		privateEndpoint := resp.PrivateEndpoint
		su.Require().NotEmpty(privateEndpoint)
		privateLinkService = privateEndpoint.PrivateLinkService
		su.Require().NotEmpty(privateLinkService)
		su.T().Logf("Private Link Service status: %s", privateLinkService.Status)
		if privateLinkService.Status == "ACTIVE" {
			su.accesspointPublicEndpointHost = resp.PublicEndpoint.Host
			su.accesspointVpcPeeringEndpointHost = resp.PrivateEndpoint.Host
			su.accesspointPrivateLinkEndpointHost = privateEndpoint.Host
			break
		}
		time.Sleep(time.Second * 60)
	}
	su.Require().NotNil(privateLinkService)
	serviceName := privateLinkService.ServiceName
	su.Require().NotEmpty(serviceName)

	su.Require().NotEmpty(su.accesspointPublicEndpointHost)
	su.Require().NotEmpty(su.accesspointVpcPeeringEndpointHost)
	su.Require().NotEmpty(su.accesspointPrivateLinkEndpointHost)

	resp := su.openApiCli.ListPrivateLinkEndpoints(su.projectID, su.accesspointClusterID, su.accesspointID)
	for _, item := range resp.Items {
		su.accesspointPrivateLinkEndpointID = lo.FromPtr(item.ID)
		su.accesspointAwsVpcEndpointName = lo.FromPtr(item.EndpointName)
		break
	}

	if su.accesspointPrivateLinkEndpointID != "" {
		su.T().
			Logf("Skip creating private link endpoint, privateLinkEndpointID: %s", su.accesspointPrivateLinkEndpointID)
		return
	}

	// create vpc endpoint
	su.accesspointAwsVpcEndpointName = su.createAwsPrivateLink(serviceName)
	su.Require().NotEmpty(su.accesspointAwsVpcEndpointName)
	su.T().Logf("AWS private link vpcEndpointID: %s", su.accesspointAwsVpcEndpointName)

	// create private link endpoint
	su.accesspointPrivateLinkEndpointID = su.openApiCli.CreatePrivateLinkEndpoint(
		su.projectID,
		su.accesspointClusterID,
		su.accesspointID,
		su.accesspointAwsVpcEndpointName,
	)
	su.Require().NotEmpty(su.accesspointPrivateLinkEndpointID)
	su.T().Logf("accesspointPrivateLinkEndpointID: %s", su.accesspointPrivateLinkEndpointID)

	// check status of private link endpoint
	retryErr := retry.Do(
		func() error {
			resp := su.openApiCli.GetPrivateLinkEndpoint(
				su.projectID,
				su.accesspointClusterID,
				su.accesspointID,
				su.accesspointPrivateLinkEndpointID,
			)
			su.T().Logf("Private link endpoint status: %s", lo.FromPtr(resp.Status))
			if lo.FromPtr(resp.Status) == "ACTIVE" {
				return nil
			}
			return fmt.Errorf("private link endpoint status: %s", lo.FromPtr(resp.Status))
		},
		retry.Attempts(90),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(10*time.Second),
	)
	su.Require().NoError(retryErr)

	// modify vpc endpoint to enable private dns
	su.createAwsPrivateDns(su.accesspointAwsVpcEndpointName)

	// wait for private link endpoint dns ready
	hostCmd := fmt.Sprintf("host %s", su.accesspointPrivateLinkEndpointHost)
	su.T().Logf("hostCmd: %s", hostCmd)
	retryErr = retry.Do(
		func() error {
			hostCmdOutput, err := su.runCommandOnAwsEc2(hostCmd)
			su.T().Logf("hostCmdOutput: %s, err: %v", hostCmdOutput, err)
			return err
		},
		retry.Attempts(10),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(60*time.Second),
	)
	su.Require().NoError(retryErr)
	su.T().Logf("private link endpoint dns is ready")
}

func (su *TestDedicatedMigrationSuite) runWorkloadAWS() {
	if host := su.privateLinkEndpointHost; host != "" {
		runWorkloadCmd := fmt.Sprintf(
			"nohup /home/ec2-user/scripts/privatelink-workload.sh %s %s > /tmp/privatelink-%s.out 2>&1 & echo $!",
			host,
			su.password,
			su.clusterID,
		)
		su.T().Logf("runPrivatelinkWorkload: %s", runWorkloadCmd)
		var (
			output string
		)
		retryErr := retry.Do(
			func() error {
				var err error
				output, err = su.runCommandOnAwsEc2(runWorkloadCmd)
				su.T().Logf("run privatelink workload output: %s", output)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
		su.Require().NotEmpty(output)
		su.privateLinkEndpointWorkloadPID = output
	}
	if host := su.vpcPeeringEndpointHost; host != "" {
		runWorkloadCmd := fmt.Sprintf(
			"nohup /home/ec2-user/scripts/vpcpeering-workload.sh %s %s > /tmp/vpcpeering-%s.out 2>&1 & echo $!",
			host,
			su.password,
			su.clusterID,
		)
		su.T().Logf("runVPCPeeringWorkload: %s", runWorkloadCmd)
		var (
			output string
		)
		retryErr := retry.Do(
			func() error {
				var err error
				output, err = su.runCommandOnAwsEc2(runWorkloadCmd)
				su.T().Logf("run vpcpeering workload output: %s", output)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
		su.Require().NotEmpty(output)
		su.vpcPeeringEndpointWorkloadPID = output
	}
	if host := su.publicEndpointHost; host != "" {
		runWorkloadCmd := fmt.Sprintf(
			"nohup /home/ec2-user/scripts/publicendpoint-workload.sh %s %s > /tmp/publicendpoint-%s.out 2>&1 & echo $!",
			host,
			su.password,
			su.clusterID,
		)
		su.T().Logf("runPublicEndpointWorkload: %s", runWorkloadCmd)
		var (
			output string
		)

		retryErr := retry.Do(
			func() error {
				var err error
				output, err = su.runCommandOnAwsEc2(runWorkloadCmd)
				su.T().Logf("run publicendpoint workload output: %s", output)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
		su.Require().NotEmpty(output)
		su.publicEndpointWorkloadPID = output
	}
}

func (su *TestDedicatedMigrationSuite) runAccesspointWorkloadAWS() {
	if host := su.accesspointPrivateLinkEndpointHost; host != "" {
		runWorkloadCmd := fmt.Sprintf(
			"nohup /home/ec2-user/scripts/accesspoint-privatelink-workload.sh %s %s > /tmp/privatelink-%s.out 2>&1 & echo $!",
			host,
			su.accesspointClusterPassword,
			su.accesspointClusterID,
		)
		su.T().Logf("runAccesspointPrivatelinkWorkload: %s", runWorkloadCmd)
		var (
			output string
		)
		retryErr := retry.Do(
			func() error {
				var err error
				output, err = su.runCommandOnAwsEc2(runWorkloadCmd)
				su.T().Logf("run accesspoint privatelink workload output: %s", output)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
		su.Require().NotEmpty(output)
		su.accesspointPrivateLinkEndpointWorkloadPID = output
	}
	if host := su.accesspointVpcPeeringEndpointHost; host != "" {
		runWorkloadCmd := fmt.Sprintf(
			"nohup /home/ec2-user/scripts/accesspoint-vpcpeering-workload.sh %s %s > /tmp/vpcpeering-%s.out 2>&1 & echo $!",
			host,
			su.accesspointClusterPassword,
			su.accesspointClusterID,
		)
		su.T().Logf("runAccesspointVPCPeeringWorkload: %s", runWorkloadCmd)
		var (
			output string
		)
		retryErr := retry.Do(
			func() error {
				var err error
				output, err = su.runCommandOnAwsEc2(runWorkloadCmd)
				su.T().Logf("run accesspoint vpcpeering workload output: %s", output)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
		su.Require().NotEmpty(output)
		su.accesspointVpcPeeringEndpointWorkloadPID = output
	}
	if host := su.accesspointPublicEndpointHost; host != "" {
		runWorkloadCmd := fmt.Sprintf(
			"nohup /home/ec2-user/scripts/accesspoint-publicendpoint-workload.sh %s %s > /tmp/publicendpoint-%s.out 2>&1 & echo $!",
			host,
			su.accesspointClusterPassword,
			su.accesspointClusterID,
		)
		su.T().Logf("runAccesspointPublicEndpointWorkload: %s", runWorkloadCmd)
		var (
			output string
		)

		retryErr := retry.Do(
			func() error {
				var err error
				output, err = su.runCommandOnAwsEc2(runWorkloadCmd)
				su.T().Logf("run accesspoint publicendpoint workload output: %s", output)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
		su.Require().NotEmpty(output)
		su.accesspointPublicEndpointWorkloadPID = output
	}
}

func (su *TestDedicatedMigrationSuite) runWorkloadGCP() {
	{
		runPrivatelinkWorkloadCmd := fmt.Sprintf(
			"nohup /home/vm-user/scripts/privatelink-workload.sh %s %s > /tmp/privatelink-%s.out 2>&1 & echo $!",
			su.privateLinkEndpointHost,
			su.password,
			su.clusterID,
		)
		su.T().Logf("runPrivatelinkWorkload: %s", runPrivatelinkWorkloadCmd)
		var (
			output string
		)
		retryErr := retry.Do(
			func() error {
				var err error
				output, err = su.runCommandOnGcpPrivateLinkVm(runPrivatelinkWorkloadCmd)
				su.T().Logf("run privatelink workload output: %s", output)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
		su.Require().NotEmpty(output)
	}
	{
		runVPCPeeringWorkloadCmd := fmt.Sprintf(
			"nohup /home/vm-user/scripts/vpcpeering-workload.sh %s %s > /tmp/vpcpeering-%s.out 2>&1 & echo $!",
			su.vpcPeeringEndpointHost,
			su.password,
			su.clusterID,
		)
		su.T().Logf("runVPCPeeringWorkload: %s", runVPCPeeringWorkloadCmd)
		var (
			output string
		)
		retryErr := retry.Do(
			func() error {
				var err error
				output, err = su.runCommandOnGcpVpcPeeringVm(runVPCPeeringWorkloadCmd)
				su.T().Logf("run vpcpeering workload output: %s", output)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
		su.Require().NotEmpty(output)
	}
	{
		runPublicEndpointWorkloadCmd := fmt.Sprintf(
			"nohup /home/vm-user/scripts/publicendpoint-workload.sh %s %s > /tmp/publicendpoint-%s.out 2>&1 & echo $!",
			su.publicEndpointHost,
			su.password,
			su.clusterID,
		)
		su.T().Logf("runPublicEndpointWorkload: %s", runPublicEndpointWorkloadCmd)
		var (
			output string
		)

		retryErr := retry.Do(
			func() error {
				var err error
				output, err = su.runCommandOnGcpVpcPeeringVm(runPublicEndpointWorkloadCmd)
				su.T().Logf("run publicendpoint workload output: %s", output)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
		su.Require().NotEmpty(output)
	}
}

func (su *TestDedicatedMigrationSuite) Test_012_RunWorkload() {
	switch su.provider {
	case "aws":
		su.runWorkloadAWS()
		if !su.skipAccesspoint {
			su.runAccesspointWorkloadAWS()
		}

	case "gcp":
		su.runWorkloadGCP()
	}
}

func (su *TestDedicatedMigrationSuite) Test_013_WaitForClusterPaused() {
	if su.skipPauseClusterCase {
		su.T().Logf("Skip pausing cluster, clusterID: %s", su.pausedClusterIDBeforeMig)
		return
	}

	su.waitForClusterStatusNormal(
		model.CentralClusterStatusPAUSED,
		su.pausedClusterIDBeforeMig,
		su.pausedClusterNameBeforeMig,
	)
}

type InstanceQueryItem struct {
	Metric map[string]string `json:"metric"`
	Value  []any             `json:"value"`
}

type InstanceQueryData struct {
	ResultType string              `json:"resultType"`
	Result     []InstanceQueryItem `json:"result"`
}
type InstanceQueryStats struct {
	SeriesFetched     string `json:"seriesFetched"`
	ExecutionTimeMsec int    `json:"executionTimeMsec"`
}

type InstanceQueryResp struct {
	Status    string             `json:"status"`
	IsPartial bool               `json:"isPartial"`
	Data      InstanceQueryData  `json:"data"`
	Stats     InstanceQueryStats `json:"stats"`
}

// call o11y ds /external/metrics/{biz_type}/tenant/{org_id}/project/{project_id}/application/{cluster_id}/api/v1/query_range,GET
// to get dataplane pod uptime
func (su *TestDedicatedMigrationSuite) getProcessStartTimeMillSeconds() (map[string]uint64, error) {
	o11yHost := fmt.Sprintf("https://www.ds.%s.%s.observability.tidbcloud.com", su.region, su.provider)
	query := make(url.Values)
	query.Add(
		"query",
		fmt.Sprintf( //nolint: gocritic // we use back quote to keep the format
			`label_keep(100*process_start_time_seconds{tidb_cluster_id="%s",component!="auditlog"},"instance")`,
			su.clusterID,
		),
	)
	o11yApplicationID := fmt.Sprintf("%s-tidbcloud", su.env)
	if su.env == "prod" {
		o11yApplicationID = "tidbcloud"
	}
	u := fmt.Sprintf("%s/external/metrics/%s/tenant/%s/project/%s/application/%s/api/v1/query?%s",
		o11yHost,
		o11yApplicationID,
		su.orgID,
		su.projectID,
		su.clusterID,
		query.Encode())
	su.T().Log(u)
	req, err := http.NewRequest(http.MethodGet, u, http.NoBody)
	if err != nil {
		return nil, err
	}
	req.Header.Add("authorization",
		"Bearer "+su.BearerToken)

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		su.T().Logf("do request failed, err:%v", err)
		return nil, err
	}
	defer resp.Body.Close()
	data, err := io.ReadAll(resp.Body)
	if err != nil {
		su.T().Logf("read body failed, err:%v", err)
		return nil, err
	}
	if resp.StatusCode > 300 {
		return nil, errors.New(string(data))
	}
	result := new(InstanceQueryResp)
	if err := json.Unmarshal(data, result); err != nil {
		return nil, err
	}
	out := map[string]uint64{}
	for _, v := range result.Data.Result {
		value, ok := v.Value[1].(string)
		if !ok {
			return nil, errors.New("cannot parse instance query result")
		}
		ts, err := strconv.ParseUint(value, 10, 64)
		if err != nil {
			return nil, err
		}
		instance := v.Metric["instance"]
		out[instance] = ts
	}
	return out, err
}

func (su *TestDedicatedMigrationSuite) Test_014_RecordPodUpTime() {
	out, err := su.getProcessStartTimeMillSeconds()
	su.Require().NoError(err)
	su.podStartTimestampBeforeMig = out
}

func (su *TestDedicatedMigrationSuite) Test_015_CheckO11yData() {
	if su.skipO11yCase {
		su.T().Logf("Skip checking o11y case")
		return
	}

	ctx := context.Background()
	now := int(time.Now().Unix())
	su.checkO11yData(ctx, su.clusterID, now-60*60, now)
}