kubernetes控制平面组件：etcd

发布一下 0 0

搭建etcd集群

机制

  • 静态
  • etcd动态发现
  • DNS发现

常见参数

--listen-peer-urls

用于监听伙伴通讯的URL列表。这个标记告诉 etcd 在特定的 scheme://IP:port 组合上从它的伙伴接收进来的请求，http 或者 https

环境变量: ETCD_LISTEN_PEER_URLS

--listen-client-urls

用于监听客户端通讯的URL列表。这个标记告诉 etcd 在特定的 scheme://IP:port 组合上从客户端接收进来的请求。scheme 可是 http 或者 https。

环境变量: ETCD_LISTEN_CLIENT_URLS

--initial-advertise-peer-urls

列出这个成员的伙伴 URL 以便通告给集群的其他成员。这些地方用于在集群中通讯 etcd 数据。至少有一个必须对所有集群成员可以路由的。这些 URL 可以包含域名。

环境变量: ETCD_INITIAL_ADVERTISE_PEER_URLS

--initial-cluster

为启动初始化集群配置。

环境变量: ETCD_INITIAL_CLUSTER

--initial-cluster-state

初始化集群状态("new" or "existing")。在初始化静态(initial static)或者 DNS 启动 (DNS bootstrapping) 期间为所有成员设置为 new 。如果这个选项被设置为 existing , etcd 将试图加入已有的集群。如果设置为错误的值，etcd 将尝试启动但安全失败。

环境变量: ETCD_INITIAL_CLUSTER_STATE

--advertise-client-urls

列出这个成员的客户端URL，通告给集群中的其他成员。这些 URL 可以包含域名。

环境变量: ETCD_ADVERTISE_CLIENT_URLS


图形化配置

http://play.etcd.io/install


在kubernetes里部署

https://bitnami.com/stack/etcd/helm

1.code

https://gitee.com/zhouguanyulovejiadanyang/mygit/blob/k8s/sts.yamlhttps://github.com/kubernetes/kubernetes/tree/master/test/e2e/testing-manifests/statefulset/etcd

headless svc, 像DNS RR ClusterIP:None

kubectl -n stg1 get endpoints

apiVersion: v1 kind: Service metadata:   name: etcd   labels:     app: etcd spec:   ports:   - port: 2380     name: etcd-server   - port: 2379     name: etcd-client   clusterIP: None   selector:     app: etcd   publishNotReadyAddresses: true

client 怎么访问：

kubectl port-forward svc/etcd-cluster-client 2379:2379


apiVersion: v1kind: Servicemetadata:  labels:    app: etcd  name: etcd-cluster-clientspec:  ports:  - name: etcd-cluster-2379    port: 2379    protocol: TCP    targetPort: 2379  selector:    app: etcd  sessionAffinity: None  type: ClusterIP


2.配置文件

  • PDB
  • service
  • sts

3.apply


官方的code有两个问题

  • etcd版本老
  • 配置不对： 2022-01-29 14:47:51.344590 E | etcdmain: error verifying flags, expected IP in URL for binding (http://etcd-0.etcd:2380). See 'etcd --help' ，暂时不要用3.4，3.5的 image

本地访问

kubectl port-forward svc/etcd-cluster-client 2379:2379

扩容

~/etcd ❯❯❯ kubectl scale --replicas=5 sts etcdstatefulset.apps/etcd scaled~/etcd ❯❯❯ kubectl get podNAME     READY   STATUS    RESTARTS   AGEetcd-0   1/1     Running   0          12metcd-1   1/1     Running   0          12metcd-2   1/1     Running   0          12metcd-3   0/1     Pending   0          2s

利用反亲和性 分布etcd pod到不同节点

https://kubernetes.io/zh/docs/concepts/scheduling-eviction/assign-pod-node/

常用命令

  • 集群状态
/ # export ETCDCTL_API=3/ # etcdctl --endpoints etcd-0.etcd:2379,etcd-1.etcd:2379,etcd-2.etcd:2379 endpoint status --write-out=table+------------------+------------------+---------+---------+-----------+-----------+------------+|     ENDPOINT     |        ID        | VERSION | DB SIZE | IS LEADER | RAFT TERM | RAFT INDEX |+------------------+------------------+---------+---------+-----------+-----------+------------+| etcd-0.etcd:2379 | c799a6ef06bc8c14 |   3.3.8 |   20 kB |     false |         6 |          9 || etcd-1.etcd:2379 | 9869f0647883a00d |   3.3.8 |   20 kB |      true |         6 |          9 || etcd-2.etcd:2379 | 42c8b94265b9b79a |   3.3.8 |   20 kB |     false |         6 |          9 |+------------------+------------------+---------+---------+-----------+-----------+------------+
  • 查看member
bash-3.2$ etcdctl member list42c8b94265b9b79a, started, etcd-2, http://etcd-2.etcd.default.svc.cluster.local:2380, http://etcd-2.etcd.default.svc.cluster.local:2379, false58996ec4dced38b4, started, etcd-4, http://etcd-4.etcd.default.svc.cluster.local:2380, http://etcd-4.etcd.default.svc.cluster.local:2379, false9869f0647883a00d, started, etcd-1, http://etcd-1.etcd.default.svc.cluster.local:2380, http://etcd-1.etcd.default.svc.cluster.local:2379, falseb4476a6cf0b4d72c, started, etcd-3, http://etcd-3.etcd.default.svc.cluster.local:2380, http://etcd-3.etcd.default.svc.cluster.local:2379, falsec799a6ef06bc8c14, started, etcd-0, http://etcd-0.etcd.default.svc.cluster.local:2380, http://etcd-0.etcd.default.svc.cluster.local:2379, false
  • 健康检查
bash-3.2$ etcdctl endpoint health127.0.0.1:2379 is healthy: successfully committed proposal: took = 1.051829439s
  • 增删改查
bash-3.2$ etcdctl put /test/key "fmeng"OK
bash-3.2$ etcdctl get /test/key/test/keyfmeng
bash-3.2$ etcdctl put /test/key "fmeng is dashuaib"OK
bash-3.2$ etcdctl get /test/key/test/keyfmeng is dashuaib

~ ❯❯❯ etcdctl get / --prefix

  • watch
bash-3.2$ etcdctl watch /test/key --rev=1PUT/test/keyfmengPUT/test/keyfmeng is dashuaib

备份与还原

# 备份命令$ export ETCDCTL_API=3 $ etcdctl --endpoints=${ENDPOINTS} snapshot save /data/etcd_backup_dir/etcd-snapshot.db$ etcdctl --endpoints=${ENDPOINTS} snapshot restore snapshot.db


kubernetes控制平面组件：etcd

原理

架构

kubernetes控制平面组件：etcd

从 etcd 的架构图中我们可以看到，etcd 主要分为四个部分。

  • HTTP Server：用于处理用户发送的 API 请求以及其它 etcd 节点的同步与心跳信息请求。
  • Store：用于处理 etcd 支持的各类功能的事务，包括数据索引、节点状态变更、监控与反馈、事件处理与执行等等，是 etcd 对用户提供的大多数 API 功能的具体实现。
  • Raft：Raft 强一致性算法的具体实现，是 etcd 的核心。
  • WAL：Write Ahead Log（预写式日志），是 etcd 的数据存储方式。除了在内存中存有所有数据的状态以及节点的索引以外，etcd 就通过 WAL 进行持久化存储。WAL 中，所有的数据提交前都会事先记录日志。Snapshot 是为了防止数据过多而进行的状态快照；Entry 表示存储的具体日志内容。


kubernetes控制平面组件：etcd

  • etcd面向client和peer节点开放http服务以及grpc服务，对于像watch机制就是基于grpc的stream通信模式实现的；
  • EtcdServer是etcd上层结构体，其负责对外提供服务，且负责应用层的实现，比如操作应用层存储器，管理leassor、watch；
  • raftNode负责上层与raft层的衔接。其负责将应用的需求传递到raft中进行处理（通过Step函数）、在消息发送到其他节点前将消息保存到WAL中、调用传输器发送消息；
  • raft是raft协议的承载者；
  • raftLog用于存储状态机信息：memoryStorge保存稳定的记录，unstable保存不稳定的记录。

v2/v3

etcd 目前支持 V2 和 V3 两个大版本，这两个版本在实现上有比较大的不同，一方面是对外提供接口的方式，另一方面就是底层的存储引擎，V2 版本的实例是一个纯内存的实现，所有的数据都没有存储在磁盘上，而 V3 版本的实例就支持了数据的持久化。

v3默认boltdb

https://github.com/boltdb/bolthttps://github.com/boltdb/boltdhttps://dbdb.io/db/boltdb

consortium etcd2+mysql

存储

数据默认会存放在 /var/lib/etcd/default/ 目录。我们会发现数据所在的目录，会被分为两个文件夹中，分别是 snap 和 wal目录。

  • snap
  1. 存放快照数据，存储etcd的数据状态
  2. etcd防止WAL文件过多而设置的快照
  • wal
  1. 存放预写式日志
  2. 最大的作用是记录了整个数据变化的全部历程
  3. 在etcd中，所有数据的修改在提交前都要先写入到WAL中

Raft算法

解决三个问题：节点选举、日志复制以及安全性。

每一个 Raft 集群中都包含多个服务器，在任意时刻，每一台服务器只可能处于 Leader、Follower 以及 Candidate 三种状态；在处于正常的状态时，集群中只会存在一个 Leader 状态，其余的服务器都是 Follower 状态。

所有的 Follower 节点都是被动的，它们不会主动发出任何的请求，只会响应 Leader 和 Candidate 发出的请求。对于每一个用户的可变操作，都会被路由给 Leader 节点进行处理，除了 Leader 和 Follower 节点之外，Candidate 节点其实只是集群运行过程中的一个临时状态。

每一个服务器都会存储当前集群的最新任期，它就像是一个单调递增的逻辑时钟，能够同步各个节点之间的状态，当前节点持有的任期会随着每一个请求被传递到其他的节点上。Raft 协议在每一个任期的开始时都会从一个集群中选出一个节点作为集群的 Leader 节点，这个节点会负责集群中的日志的复制以及管理工作。

https://zhuanlan.zhihu.com/p/32052223
kubernetes控制平面组件：etcd

Watch

客户端通过监听指定的key可以迅速感知key的变化并作出相应处理，watch机制的实现依赖于资源版本号revision的设计，每一次key的更新都会使得revision原子递增，因此根据不同的版本号revision的对比就可以感知新事件的发生。etcd watch机制有着广泛的应用，比如利用etcd实现分布式锁；k8s中监听各种资源的变化，从而实现各种controller逻辑等。


kubernetes控制平面组件：etcd

watch机制的实现主要可分为三个部分

  • 客户端gRPC调用
  • server端gRPC处理
  • 从底层存储获取更新事件

client使用watchClient的watch接口发起watch请求，与server端建立一个gRPCStream连接。

server端会为每个client生成唯一一个watch id，并记录每个client也就是watcher监听的key或者key range，通过recvLoop接收client请求，通过sendLoop发送请求，server端只负责收发请求和响应。

主要的实现都放在了watchalbStore层，watchalbStore会监听key的变化，然后通过syncWatchersLoop和syncVictimsLoop两个处理流程将key的更新变化包装成event，通过channel发送给gRPC server。

MVCC

MVCC（Multiversion Concurrency Control）多版本并发控制机制

  1. 多版本
bash-3.2$ etcdctl put /奥特曼 "他很帅"OKbash-3.2$bash-3.2$ etcdctl put /奥特曼 "他真的很帅吗?"OKbash-3.2$ etcdctl del /奥特曼1bash-3.2$ etcdctl watch /奥特曼 --rev=1PUT/奥特曼他很帅PUT/奥特曼他真的很帅吗?DELETE/奥特曼

多版本的本质就是把一个key的历史变化都存下来

多版本是watch机制的基石


  1. 并发控制

场景1:

程序A：我要修改 fmeng 这个key的value数据库：好，你改吧，我加锁。程序B：我也要修改 fmeng数据库：你等着，等A改好，你再来。具体看我的锁什么时候释放。

这就是悲观锁

悲观锁：悲观得认为并发事务会冲突，所以要先拿锁，拿到锁的作修改操作

  • 读写锁（可多读，写时互斥）、互斥锁（线程互斥）等
  • 控制粒度太大，高并发下大量事务会被阻塞。

场景2

程序A：我要修改 fmeng 这个key的value数据库：好的，你修改value时候请将版本号设置为12程序A：改好了，帮我写回磁盘程序B：我修改 fmeng 的value数据库：好的，你修改value时候请将版本号设置为13程序B：改好了，帮我写回磁盘同时程序C：我修改 fmeng 的value数据库：好的，你修改value时候请将版本号设置为13程序C：改好了，帮我写回磁盘

数据库：写回磁盘，A写好了。哎，B和C都是version 13，我咋写？算了，报错吧。。

就是乐观锁，默认不加锁，你尽管写，冲突我认怂！乐观锁其实不是锁，只是相对悲观锁来定义，适合读多写少。

乐观锁：乐观得认为数据不会冲突，但发生冲突时要能检测到。


场景3

程序A：我要更新 fmeng数据库：你干吧，你的版本是20程序B：我要更新 fmeng数据库：你干吧，你的版本是22程序C：我要更新 fmeng数据库：你干吧，你的版本是23


这就是MVCC，在 MVCC 数据库中，你更新一个 key-value 数据的时候，它并不会直接覆盖原数据，而是新增一个版本来存储新的数据，每个数据都有一个版本号，版本号是一个逻辑时钟，不会因为服务器时间的差异而受影响。

MVCC不等于乐观锁！

https://juejin.cn/post/6885290108480651272
  1. etcd中的MVCC
https://pkg.go.dev/go.etcd.io/etcd/server/v3/mvcctype revision struct {    main int64    // 一个全局递增的主版本号，随put/txn/delete事务递增，一个事务内的key main版本号是一致的  sub int64    // 一个事务内的子版本号，从0开始随事务内put/delete操作递增}

--rev 查的是main

Goetcdctl watch /奥特曼 --rev=1

在底层boltdb里，实际分布是这样的：

kubernetes控制平面组件：etcd

底层的key是revision，/奥特曼是用户key，“他很帅”就是用户value

删除

之前有delete动作，但是依然有版本记录。为什么？

etcdctl del /奥特曼

删除这个动作，其实etcd是在blotdb里写了一条，“删除用户/奥特曼”

此时有个问题：用户说我的确删除了啊，真的不要了！请把空间还给我啊！

回收 compact（压缩）

为了防止空间不够用，必须定期释放一些用户已经声明删除的数据，这个动作就叫做 compact。

etcdctl compact {version}

compact 需要一个版本号。这个版本号就是写事务递增的那个版本号，compact 12345，就是说把版本12345以前的标记删除了的数据释放掉，用户没删除的数据肯定不能回收。

如何压缩：

  • etcd 可以使用带有小时时间单位的 --auto-compaction 选项来设置为自动压缩键空间:
  • etcdctl 如下发起压缩工作: etcdctl compact 123456


开发篇

CRUD

注意修改go.mod

replace github.com/coreos/bbolt => go.etcd.io/bbolt v1.3.6replace google.golang.org/grpc => google.golang.org/grpc v1.26.0
package mainimport (   "context"   "fmt"   "time"   "go.etcd.io/etcd/clientv3")func main() {     // init client   cli, err := clientv3.New(clientv3.Config{   Endpoints:   []string{"127.0.0.1:2379"},   DialTimeout: 5 * time.Second,   })   if err != nil {		fmt.Printf("connect to etcd failed, err:%v\n", err)		return	}	fmt.Println("connect to etcd success")	defer cli.Close()	// put	ctx, cancel := context.WithTimeout(context.Background(), time.Second)	_, err = cli.Put(ctx, "奥特曼", "大帅哥")	cancel()	if err != nil {		fmt.Printf("put to etcd failed, err:%v\n", err)		return	}	// get	ctx, cancel = context.WithTimeout(context.Background(), time.Second)	resp, err := cli.Get(ctx, "奥特曼")	cancel()	if err != nil {		fmt.Printf("get from etcd failed, err:%v\n", err)		return	}	for _, ev := range resp.Kvs {		fmt.Printf("%s:%s\n", ev.Key, ev.Value)	}}

Watch

package mainimport (   "context"   "fmt"   "time"   "go.etcd.io/etcd/clientv3")// watch demofunc main() {    cli, err := clientv3.New(clientv3.Config{		Endpoints:   []string{"127.0.0.1:2379"},		DialTimeout: 5 * time.Second,	})	if err != nil {		fmt.Printf("connect to etcd failed, err:%v\n", err)		return	}	fmt.Println("connect to etcd success")	defer cli.Close()	rch := cli.Watch(context.Background(), "/test/key") // <-chan WatchResponse	for wresp := range rch {		for _, ev := range wresp.Events {			fmt.Printf("Type: %s Key:%s Value:%s\n", ev.Type, ev.Kv.Key, ev.Kv.Value)		}	}}

服务发现

kubernetes控制平面组件：etcd

服务发现要解决的也是分布式系统中最常见的问题之一，即在同一个分布式集群中的进程或服务，要如何才能找到对方并建立连接。本质上来说，服务发现就是想要了解集群中是否有进程在监听 udp 或 tcp 端口，并且通过名字就可以查找和连接。

需要实现的功能；

  • 服务发现：通过服务节点能查询到服务提供外部访问的 IP 和端口号。比如网关代理服务时能够及时的发现服务中新增节点、丢弃不可用的服务节点。
  • 服务注册：同一service的所有节点注册到相同目录下，节点启动后将自己的信息注册到所属服务的目录中。
  • 健康检查：服务节点定时进行健康检查。注册到服务目录中的信息设置一个较短的TTL，运行正常的服务节点每隔一段时间会去更新信息的TTL ，从而达到健康检查效果。

discover.go

package mainimport (	"context"	"log"	"sync"	"time"	"github.com/coreos/etcd/mvcc/mvccpb"	"go.etcd.io/etcd/clientv3")//ServiceDiscovery 服务发现type ServiceDiscovery struct {	cli        *clientv3.Client  //etcd client	serverList map[string]string //服务列表	lock       sync.Mutex}//NewServiceDiscovery  新建发现服务func NewServiceDiscovery(endpoints []string) *ServiceDiscovery {	cli, err := clientv3.New(clientv3.Config{		Endpoints:   endpoints,		DialTimeout: 5 * time.Second,	})	if err != nil {		log.Fatal(err)	}	return &ServiceDiscovery{		cli:        cli,		serverList: make(map[string]string),	}}//WatchService 初始化服务列表和监视func (s *ServiceDiscovery) WatchService(prefix string) error {	//根据前缀获取现有的key	resp, err := s.cli.Get(context.Background(), prefix, clientv3.WithPrefix())	if err != nil {		return err	}	for _, ev := range resp.Kvs {		s.SetServiceList(string(ev.Key), string(ev.Value))	}	//监视前缀，修改变更的server	go s.watcher(prefix)	return nil}//watcher 监听前缀func (s *ServiceDiscovery) watcher(prefix string) {	rch := s.cli.Watch(context.Background(), prefix, clientv3.WithPrefix())	log.Printf("watching prefix:%s now...", prefix)	for wresp := range rch {		for _, ev := range wresp.Events {			switch ev.Type {			case mvccpb.PUT: //修改或者新增				s.SetServiceList(string(ev.Kv.Key), string(ev.Kv.Value))			case mvccpb.DELETE: //删除				s.DelServiceList(string(ev.Kv.Key))			}		}	}}//SetServiceList 新增服务地址func (s *ServiceDiscovery) SetServiceList(key, val string) {	s.lock.Lock()	defer s.lock.Unlock()	s.serverList[key] = string(val)	log.Println("put key :", key, "val:", val)}//DelServiceList 删除服务地址func (s *ServiceDiscovery) DelServiceList(key string) {	s.lock.Lock()	defer s.lock.Unlock()	delete(s.serverList, key)	log.Println("del key:", key)}//GetServices 获取服务地址func (s *ServiceDiscovery) GetServices() []string {	s.lock.Lock()	defer s.lock.Unlock()	addrs := make([]string, 0)	for _, v := range s.serverList {		addrs = append(addrs, v)	}	return addrs}//Close 关闭服务func (s *ServiceDiscovery) Close() error {	return s.cli.Close()}func main() {	var endpoints = []string{"localhost:2379"}	ser := NewServiceDiscovery(endpoints)	defer ser.Close()	ser.WatchService("/web/")	//ser.WatchService("/gRPC/")	for {		select {		case <-time.Tick(10 * time.Second):			log.Println(ser.GetServices())		}	}}
//resgister.gopackage mainimport (	"context"	"log"	"time"	"go.etcd.io/etcd/clientv3")//ServiceRegister 创建租约注册服务type ServiceRegister struct {	cli     *clientv3.Client //etcd client	leaseID clientv3.LeaseID //租约ID	//租约keepalieve相应chan	keepAliveChan <-chan *clientv3.LeaseKeepAliveResponse	key           string //key	val           string //value}//NewServiceRegister 新建注册服务func NewServiceRegister(endpoints []string, key, val string, lease int64) (*ServiceRegister, error) {	cli, err := clientv3.New(clientv3.Config{		Endpoints:   endpoints,		DialTimeout: 5 * time.Second,	})	if err != nil {		log.Fatal(err)	}	ser := &ServiceRegister{		cli: cli,		key: key,		val: val,	}	//申请租约设置时间keepalive	if err := ser.putKeyWithLease(lease); err != nil {		return nil, err	}	return ser, nil}//设置租约func (s *ServiceRegister) putKeyWithLease(lease int64) error {	//设置租约时间	resp, err := s.cli.Grant(context.Background(), lease)	if err != nil {		return err	}	//注册服务并绑定租约	_, err = s.cli.Put(context.Background(), s.key, s.val, clientv3.WithLease(resp.ID))	if err != nil {		return err	}	//设置续租 定期发送需求请求	leaseRespChan, err := s.cli.KeepAlive(context.Background(), resp.ID)	if err != nil {		return err	}	s.leaseID = resp.ID	log.Println(s.leaseID)	s.keepAliveChan = leaseRespChan	log.Printf("Put key:%s  val:%s  success!", s.key, s.val)	return nil}//ListenLeaseRespChan 监听 续租情况func (s *ServiceRegister) ListenLeaseRespChan() {	for leaseKeepResp := range s.keepAliveChan {		log.Println("续约成功", leaseKeepResp)	}	log.Println("关闭续租")}// Close 注销服务func (s *ServiceRegister) Close() error {	//撤销租约	if _, err := s.cli.Revoke(context.Background(), s.leaseID); err != nil {		return err	}	log.Println("撤销租约")	return s.cli.Close()}func main() {	var endpoints = []string{"localhost:2379"}	ser, err := NewServiceRegister(endpoints, "/web/node1", "localhost:8000", 5)	if err != nil {		log.Fatalln(err)	}	//监听续租相应chan	go ser.ListenLeaseRespChan()	select {	 case <-time.After(20 * time.Second):		 ser.Close()	}}

面试题

  1. etcd的raft和zookeeper的paxos的区别
  2. etcd和redis区别
  3. etcd性能优化（存储，网络，数据分离）
  4. etcd是如何实现一致性的？
  5. etcd的存储是如何实现的？
  6. etcd的watch机制是如何实现的？
  7. etcd的key过期机制是如何实现的，lease
  8. 乐观锁与悲观锁，MVCC


案例和扩展

eBay payment

https://cloud.tencent.com/developer/article/1538019

ebay kubernetes 控制面架构

kubernetes控制平面组件：etcd

https://time.geekbang.org/column/intro/100069901

问题

  • 怎么看etcd最多支持多少个watch的链接啊
  • CAP etcd放弃了哪个？

版权声明：内容来源于互联网和用户投稿 如有侵权请联系删除

本文地址：http://0561fc.cn/68768.html

上一篇手机经常充满电忘记关电源？手把手教你iPhone充满电后自动提醒

下一篇人民币被欧洲狂买，预示人民币国际化走向崛起？外汇市场措手不及

相关文章