【ETCD】ETCD Defrag底层原理分析

etcdctl  --endpoints=${ENDPOINTS} defrag --command-timeout=60s

ls -l /path-to-etcd-dir/etcd/member/snap/db

-rw------- 1 root root 2270765056 Dec 18 17:34 db

// bucket represents the on-file representation of a bucket.
// This is stored as the "value" of a bucket key. If the bucket is small enough,
// then its root page can be stored inline in the "value", after the bucket
// header. In the case of inline buckets, the "root" will be 0.
type bucket struct {
    root     pgid   // page id of the bucket's root-level page
    sequence uint64 // monotonically incrementing, used by NextSequence()
}

// Bucket represents a collection of key/value pairs inside the database.
type Bucket struct {
	*bucket
	tx       *Tx                // the associated transaction
	buckets  map[string]*Bucket // subbucket cache
	page     *page              // inline page reference
	rootNode *node              // materialized node for the root page.
	nodes    map[pgid]*node     // node cache

	// Sets the threshold for filling nodes when they split. By default,
	// the bucket will fill to 50% but it can be useful to increase this
	// amount if you know that your write workloads are mostly append-only.
	//
	// This is non-persisted across transactions so it must be set in every Tx.
	FillPercent float64
}

// server/storage/backend/backend.go
func (b *backend) defrag() error {
	
	...
	
	// 创建临时db文件
	dir := filepath.Dir(b.db.Path())
	temp, err := os.CreateTemp(dir, "db.tmp.*")
	
	...
	
	tdbp := temp.Name()
	tmpdb, err := bolt.Open(tdbp, 0600, &options)
	
	...

	dbp := b.db.Path()

	...
	
	// 开始执行 defrag 逻辑
	err = defragdb(b.db, tmpdb, defragLimit)
	...
	
	// defrag完成后将 tmp 文件重命名为 db 文件
	err = os.Rename(tdbp, dbp)
	if err != nil {
		b.lg.Fatal("failed to rename tmp database", zap.Error(err))
	}

	...
	
	return nil
}

// server/storage/backend/backend.go
func defragdb(odb, tmpdb *bolt.DB, limit int) error {
	// 新建的临时 boltdb 开启写事务
	tmptx, err := tmpdb.Begin(true)
	if err != nil {
		return err
	}
	// defrag 出现异常则事务回滚
	defer func() {
		if err != nil {
			tmptx.Rollback()
		}
	}()

	// 原来的 boltdb 开启读事务
	tx, err := odb.Begin(false)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	c := tx.Cursor()

	count := 0
	//通过 cursor 遍历原来的 boltdb 的所有 key
	for next, _ := c.First(); next != nil; next, _ = c.Next() {
		b := tx.Bucket(next)
		if b == nil {
			return fmt.Errorf("backend: cannot defrag bucket %s", string(next))
		}

		tmpb, berr := tmptx.CreateBucketIfNotExists(next)
		if berr != nil {
			return berr
		}
		tmpb.FillPercent = 0.9 // for bucket2seq write in for each

		if err = b.ForEach(func(k, v []byte) error {
			count++
			// 每个写事务更新的 key-value 超过 limit （默认10000）时，会重新开启新的写事务
			// 避免每次写事务更新的 key-value 过多，占用内存过大
			if count > limit {
				err = tmptx.Commit()
				if err != nil {
					return err
				}
				tmptx, err = tmpdb.Begin(true)
				if err != nil {
					return err
				}
				tmpb = tmptx.Bucket(next)
				tmpb.FillPercent = 0.9 // for bucket2seq write in for each

				count = 0
			}
			// 往新建的临时 boltdb 放入遍历得到的 key-value
			return tmpb.Put(k, v)
		}); err != nil {
			return err
		}
	}

	return tmptx.Commit()
}