tikv-gc-runner.go

package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"strings"
	"time"

	pd "github.com/pingcap/pd/client"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/store/tikv"
	"github.com/pingcap/tidb/store/tikv/gcworker"
	"github.com/pingcap/tidb/store/tikv/oracle"
)

var (
	pdServers   = flag.String("pd", "127.0.0.1:2379", "PD Addresses of the TiKV cluster. default: 127.0.0.1:2379")
	concurrency = flag.Int("concurrency", 4, "Concurrency of ResolveLocks. default: 4")
	gcLifetime   = flag.Duration("lifetime", time.Minute*10, "Time during which the data should be kept. default: 10m")
)

func createClient() (store kv.Storage, pdClient pd.Client) {
	driver := tikv.Driver{}
	var err error
	store, err = driver.Open(fmt.Sprintf("tikv://%s?disableGC=true", *pdServers))
	if err != nil {
		log.Fatalf("Failed to open db: %v", err)
	}

	addresses := strings.Split(*pdServers, ",")
	pdClient, err = pd.NewClient(addresses, pd.SecurityOption{})
	if err != nil {
		log.Fatalf("Failed to create pdClient: %v", err)
	}

	return
}

func main() {
	flag.Parse()
	if *concurrency <= 0 {
		log.Fatalf("invalid concurrency: %v", *concurrency)
	}
	if *gcLifetime < time.Minute * 10 {
		log.Fatalf("gc lifetime must be not less than 10m. current: %v", *gcLifetime)
	}
	time.Sleep(time.Second * 10)

	store, pd := createClient()
	physical, logical, err := pd.GetTS(context.Background())
	if err != nil {
		log.Fatalf("failed to get tso from pd: %+q", err)
	}
	currentTs := oracle.ComposeTS(physical, logical)
	currentTime := oracle.GetTimeFromTS(currentTs)
	safePointTime := currentTime.Add(-*gcLifetime)
	safePoint := oracle.ComposeTS(oracle.GetPhysical(safePointTime), 0)
	log.Printf("********** Run GC on Safepoint %v (%v) **********", safePoint, safePointTime)
	err = gcworker.RunDistributedGCJob(context.Background(), store.(tikv.Storage), pd, safePoint, "gc-worker", *concurrency)
	if err != nil {
		log.Fatalf("fc failed: %+q", err)
	}
}