TiDB 中的全局唯一 ID

1
2
3
4

作者：Ed Huang
链接：https://www.zhihu.com/question/52823076/answer/132331104
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// Generate a random cluster ID.
ts := uint64(time.Now().Unix())
clusterID := (ts << 32) + uint64(rand.Uint32())
value := typeutil.Uint64ToBytes(clusterID)

// Multiple PDs may try to init the cluster ID at the same time.
// Only one PD can commit this transaction, then other PDs can get
// the committed cluster ID.
resp, err := c.Txn(ctx).
	If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).
	Then(clientv3.OpPut(key, string(value))).
	Else(clientv3.OpGet(key)).
	Commit()

// Txn commits ok, return the generated cluster ID.
if resp.Succeeded {
	return clusterID, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

service PD {
	...
	rpc AllocID(AllocIDRequest) returns (AllocIDResponse) {}
	...
}

message AllocIDRequest {
    RequestHeader header = 1;
}

message AllocIDResponse {
    ResponseHeader header = 1;

    uint64 id = 2;
}

message RequestHeader {
    // cluster_id is the ID of the cluster which be sent to.
    uint64 cluster_id = 1;
}

message ResponseHeader {
    // cluster_id is the ID of the cluster which sent the response.
    uint64 cluster_id = 1;
    Error error = 2;
}
enum ErrorType {
    OK = 0;
    UNKNOWN = 1;
    NOT_BOOTSTRAPPED = 2;
    STORE_TOMBSTONE = 3;
    ALREADY_BOOTSTRAPPED = 4;
    INCOMPATIBLE_VERSION = 5;
    REGION_NOT_FOUND = 6;
}

message Error {
    ErrorType type = 1;
    string message = 2;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// AllocID implements gRPC PDServer.
func (s *Server) AllocID(ctx context.Context, request *pdpb.AllocIDRequest) (*pdpb.AllocIDResponse, error) {
	// 1. 校验请求数据，这里是在 pdpb.proto 中封装成 RequestHeader
	if err := s.validateRequest(request.GetHeader()); err != nil {
		return nil, err
	}

	// 2. 我们用一个 idAllocator 来分配 ID
	// We can use an allocator for all types ID allocation.
	id, err := s.idAllocator.Alloc()
	if err != nil {
		return nil, status.Errorf(codes.Unknown, err.Error())
	}

	// 3. 返回请求时的 header 和生成的 ID
	return &pdpb.AllocIDResponse{
		Header: s.header(),
		Id:     id,
	}, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// Server is the pd server.
type Server struct {

	// Server services.
	// for id allocator, we can use one allocator for
	// store, region and peer, because we just need
	// a unique ID.
	idAllocator *id.AllocatorImpl
}

func (s *Server) startServer() error {
}

1
2
3

func (s *Server) startServer() error {
	s.idAllocator = id.NewAllocatorImpl(s.client, s.rootPath, s.member.MemberValue())
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

// Allocator is the allocator to generate unique ID.
type Allocator interface {
	Alloc() (uint64, error)
}
// 步长 1000
const allocStep = uint64(1000)

// AllocatorImpl 是对 Allocator 的实现，用于分配 ID
// AllocatorImpl is used to allocate ID.
type AllocatorImpl struct {
	mu   sync.Mutex
	base uint64
	end  uint64

	// etcd client
	client   *clientv3.Client
	rootPath string
	member   string
}

// NewAllocatorImpl creates a new IDAllocator.
func NewAllocatorImpl(client *clientv3.Client, rootPath string, member string) *AllocatorImpl {
	return &AllocatorImpl{client: client, rootPath: rootPath, member: member}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

// Alloc returns a new id.
func (alloc *AllocatorImpl) Alloc() (uint64, error) {
	// 给分配增加锁，使用 defer 在函数结束时进行释放
	alloc.mu.Lock()
	defer alloc.mu.Unlock()

	// 第一次的时候 base 和 end 都为 0，所以会执行 generate()
	// 否则直接返回 alloc.base++
	if alloc.base == alloc.end {
		end, err := alloc.generate()
		if err != nil {
			return 0, err
		}

		alloc.end = end
		alloc.base = alloc.end - allocStep
	}

	alloc.base++

	return alloc.base, nil
}

func (alloc *AllocatorImpl) generate() (uint64, error) {
	// 获取要给 XXPath 分配 ID 的 key
	key := alloc.getAllocIDPath()
	// 从 etcd 中读取 key 所对应的值
	value, err := etcdutil.GetValue(alloc.client, key)
	if err != nil {
		return 0, err
	}

	var (
		cmp clientv3.Cmp
		end uint64
	)

	if value == nil {
		// create the key
		cmp = clientv3.Compare(clientv3.CreateRevision(key), "=", 0)
	} else {
		// update the key
		end, err = typeutil.BytesToUint64(value)
		if err != nil {
			return 0, err
		}

		cmp = clientv3.Compare(clientv3.Value(key), "=", string(value))
	}
	// 如果以前不存在，则 end 被赋值为 1000（分配的步长），否则，就是原有的值+步长
	end += allocStep
	// 将 uint64 转为 bytes
	value = typeutil.Uint64ToBytes(end)
	// 从 etcd 获取一个事务，然后将值提交到 etcd 中
	txn := kv.NewSlowLogTxn(alloc.client)
	leaderPath := path.Join(alloc.rootPath, "leader")
	t := txn.If(append([]clientv3.Cmp{cmp}, clientv3.Compare(clientv3.Value(leaderPath), "=", alloc.member))...)
	resp, err := t.Then(clientv3.OpPut(key, string(value))).Commit()
	if err != nil {
		return 0, err
	}
	if !resp.Succeeded {
		return 0, errors.New("generate id failed, we may not leader")
	}

	log.Info("idAllocator allocates a new id", zap.Uint64("alloc-id", end))
	idGauge.WithLabelValues("idalloc").Set(float64(end))
	return end, nil
}

func (alloc *AllocatorImpl) getAllocIDPath() string {
	return path.Join(alloc.rootPath, "alloc_id")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

newRegionID, err := c.s.idAllocator.Alloc()

...

peerIDs := make([]uint64, len(request.Region.Peers))
for i := 0; i < len(peerIDs); i++ {
	if peerIDs[i], err = c.s.idAllocator.Alloc(); err != nil {
		return nil, err
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// prometheusPushClient pushs metrics to Prometheus Pushgateway.
func prometheusPushClient(job, addr string, interval time.Duration) {
	for {
		err := push.FromGatherer(
			job, push.HostnameGroupingKey(),
			addr,
			prometheus.DefaultGatherer,
		)
		if err != nil {
			log.Error("could not push metrics to Prometheus Pushgateway", zap.Error(err))
		}

		time.Sleep(interval)
	}
}

// Push metircs in background.
func Push(cfg *MetricConfig) {
	if cfg.PushInterval.Duration == zeroDuration || len(cfg.PushAddress) == 0 {
		log.Info("disable Prometheus push client")
		return
	}

	log.Info("start Prometheus push client")

	interval := cfg.PushInterval.Duration
	go prometheusPushClient(cfg.PushJob, cfg.PushAddress, interval)
}