关于 cubefs blobstore volume chunk 的数据均衡问题

[jianwei1216]

最近在关注 cubefs 的 blobstore 的设计
读到 <CubeFS存储技术揭秘(3)— 均衡、巡检与故障自愈>

有个疑问需要诸位的答疑解惑，看看我的理解对不对

背景：

均衡：Chunk级别的数据迁移，保障集群中各个节点Chunk数相对均衡
BlobStore集群由一个个卷（volume）组织用户数据，不同的卷可能为不同的纠删码模式，一个卷由N+M个卷单元（chunk）组成（如果有本地校验块的话，则是N+M+L个卷单元），一个chunk对应磁盘上的一个文件，chunk大小是有上限的（最大为16GB），换而言之一块磁盘上的chunk数目是固定的。
磁盘上的chunk用完可以分为两种情况：

• 磁盘空间已经用完，即每个chunk均写满，这个时候对应磁盘只能提供读，无法提供写（随着业务数据的删除，删除空间被回收，又会继续提供新的写能力，但是仍然无法组织出新卷）；
• 磁盘空间未用完，但是chunk数已用完，此时磁盘还能正常提供读写，但是无法提供chunk组织出新卷。
  当出现chunk数用完的时候，是需要通过扩容的方式来解决。

疑问：磁盘空间已经用完，即每个chunk均写满，这个时候对应磁盘只能提供读，无法提供写（随着业务数据的删除，删除空间被回收，又会继续提供新的写能力，但是仍然无法组织出新卷
极端场景：

1. 所有 chunk（16G） 文件全部写满一遍
2. 每个 chunk 文件中，都删除了 15G 的数据，留存了 1G 的有效数据在 chunk 文件的尾部
3. 这个 punch hole 只能回收 chunk 文件 0-15G 的磁盘空间，而 15G-16G 的磁盘空间依旧占用着，因此文件的逻辑空间还是 16G

问题1：这个时候，这些 16G 的 chunk 文件还能够继续写吗？

• 我理解是不能够继续写了，
• 因为采用了 append-only 写的语义，chunk 文件的逻辑空间为 16G（实际1G磁盘占用）
• 此时，再继续追加写，会超出 16G 的 chunk size 限制

问题2: 如果有一部分 chunk 被强制预留出来腾挪空间，能否解决这个问题？

• 将 volume (ec3+3) 绑定的 6 个 chunk 文件，冻结后，6 个 chunk 文件同时进行 chunk 数据迁移
• 即，将逻辑空间 16G（实际1G）的 chunk，腾挪迁移到空闲的 chunk 中，offset 从 0 开始追加写
• 理论上，此时 volume 绑定的 6 个 chunk 全部都变为逻辑空间 1G （实际也空间1G）了
• 这个时候，这个曾经被写满的 volume，还能够继续分配给 access 承接用户数据吗？
• blobstore volume 允许从写满变为可写的状态转换吗？

问题3：volume 不回收， chunk 能够回收吗？

• 场景

  • ec(3+3)
  • volume-id1 <==> [chunk1, 2, 3, 4, 5, 6] ==chunk迁移==> [chunk1, 2, 3, 4, 5, 19]
  • volume-id2 <==> [chunk7, 8, 9 , 10 ,11, 12]
  • volume-id3 <==> [chunk13, 14, 15, 16, 17, 18]

• 疑问

  • chunk 迁移，只是为了 BlobNode 节点间的 chunk 数量均衡
  • blob 被删除后，BlobNode 上的 bid ==> chunkid 中的物理磁盘块会被文件系统 punch hole
    • chunk 文件被 punch hole 后，如果一个 chunk 文件写满后 8G
    • 即使一个 chunk 文件中只有 1 G 的有效数据，剩余的 7G 都是空洞
    • 问题4 ：那这个 chunk 对应的 blobstore volume 还能够继续写入数据吗???
  • 问题5: volume-id 不回收， chunk 数量就不会变少 （这个理解对不对？）
  • 问题6: volume-id 要回收
    • chunk 数量才会减少
    • 应用存储的 location 位置就要修改 blobid <==> volume-id 的映射关系

@pengtianyue025 @sejust @true1064

[Cloudstriff]

问题1：可以继续写入，append-only语义是针对blob对象的，volume本身只要底下chunk有空间就可以继续写入
问题2：blobnode自身会预留空间，不过预留的空间主要目的是为了做chunk compaction用，而不是为了解决问题1
问题3-6：volume这个抽象设计的目的是作为数据写入的容器，本身就是为了支撑写入，只要有空间，volume就可以持续写入，因此并不需要做volume级别的compaction，底层chunk做垃圾回收即可。理解这点的话，3-6的疑问就可以自然解决了

[jianwei1216]

我读了 blobnode 这部分的源代码

这句话是正确的：Chunk 形式上是一个个大文件，大小有上限，Chunk只支持Append写，对HDD友好。

====================================================================

volume这个抽象设计的目的是作为数据写入的容器，本身就是为了支撑写入，只要有空间，volume就可以持续写入，因此并不需要做volume级别的compaction，底层chunk做垃圾回收即可。

blobstore volume 应该不只是一个容器 ，它同样也是一个追加写带 EC (n+m) 的大文件

===================================================================
疑问1：Chunk 大小有上限，这里的大小是指文件的逻辑空间，还是指实际占用的物理空间？

====================================================================
疑问2："如果疑问1是指 chunk 文件的逻辑空间的话"，punch hole 可以立即让本地文件系统回收磁盘空间，对于已经写满的 chunk file(volume)， blobstore volume （n+m chunkfiles) 只有通过迁移 chunk，才能够让一个写满的 volume，重新变的可写

比如：
EC 3+3
blobstore volume-1 [chunk1,2,3,4,5,6] ==迁移到==> [chunk7,8,9,10,11,12]

====================================================================
下面是写 shard 的源码分析

1. blobnode client 发送 PutShard

//blobstore/api/blobnode/shard.go
func (c *client) PutShard(ctx context.Context, host string, args *PutShardArgs) (crc uint32, err error) {
    if args.Size > MaxShardSize {
        err = bloberr.ErrShardSizeTooLarge
        return
    }
    if !args.Type.IsValid() {
        err = bloberr.ErrInvalidParam
        return
    }

    if !IsValidDiskID(args.DiskID) {
        err = bloberr.ErrInvalidDiskId
        return
    }

    ret := &PutShardRet{
        Crc: proto.InvalidCrc32,
    }
    urlStr := fmt.Sprintf("%v/shard/put/diskid/%v/vuid/%v/bid/%v/size/%v?iotype=%d",
        host, args.DiskID, args.Vuid, args.Bid, args.Size, args.Type)
    req, err := http.NewRequest(http.MethodPost, urlStr, args.Body)
    if err != nil {
        return
    }
    req.ContentLength = args.Size
    err = c.DoWith(ctx, req, ret, rpc.WithCrcEncode())
    if err == nil {
        crc = ret.Crc
    }

    return
}

2. blobnode server 收到 PutShard 请求后，调用 ShardPut

// blobstore/blobnode/service.go
func NewHandler(service *Service) *rpc.Router {
    r := rpc.New()

    rpc.RegisterArgsParser(&bnapi.PutShardArgs{}, "json")
    r.Handle(http.MethodPost, "/shard/put/diskid/:diskid/vuid/:vuid/bid/:bid/size/:size", service.ShardPut, rpc.OptArgsURI(), rpc.OptArgsQuery())
}

3. blobnode server ShardPut 调用 Chunk Write

//blobstore/blobnode/shard.go
/*
 *  method:         POST
 *  url:            /shard/put/diskid/{diskid}/vuid/{vuid}/bid/{bid}/size/{size}?iotype={iotype}
 *  request body:   bidData
 */
func (s *Service) ShardPut(c *rpc.Context) {
    args := new(bnapi.PutShardArgs)

    err := cs.AllowModify()
    shard := core.NewShardWriter(args.Bid, args.Vuid, uint32(args.Size), c.Request.Body)
    err = cs.Write(ctx, shard)
}

4. blobnode server Chunk Write 调用 storage Write

/*
Need Shard:
    - Size
    - Flag
    - Body (Reader)
Fill Shard:
    - Offset
    - Crc
*/
func (cs *chunk) Write(ctx context.Context, b *core.Shard) (err error) {
    if b.Vuid != cs.vuid {
        return bloberr.ErrVuidNotMatch
    }

    elem := cs.consistent.Begin(b.Bid)
    defer cs.consistent.End(elem)

    // statistics
    cs.stats.writeBefore()
    defer cs.stats.writeAfter(uint64(b.Size), time.Now())

    cs.lock.RLock()

    if cs.compacting {
        cs.bidlimiter.Acquire(b.Bid)
        defer cs.bidlimiter.Release(b.Bid)
    }
    stg := cs.GetStg()
    defer cs.PutStg(stg)

    cs.lock.RUnlock()

    if err = stg.Write(ctx, b); err != nil {
        return err
    }

    // update stats
    atomic.AddUint64(&cs.fileInfo.Used, uint64(core.Alignphysize(int64(b.Size))))
    atomic.StoreUint32(&cs.dirty, 1)

    return nil
}

5. blobnode server storage Write 调用 datafile Write

func (stg *storage) Write(ctx context.Context, b *core.Shard) (err error) {
    data, meta := stg.data, stg.meta

    // write data file, will modify *shard
    err = data.Write(ctx, b)
    if err != nil {
        return err
    }

    // write meta
    return meta.Write(ctx, b.Bid, core.ShardMeta{
        Version: _shardVer[0],
        Size:    b.Size,
        Crc:     b.Crc,
        Offset:  b.Offset,
        Flag:    b.Flag,
    })
}

6. datafile Write 调用 allocSpace 分配 offset （pos）写入位置，分别写入 header/body/footer

func (cd *datafile) Write(ctx context.Context, shard *core.Shard) error {
	span := trace.SpanFromContextSafe(ctx)

	var (
		w      *bncomm.Writer
		buffer []byte
		start  time.Time
	)

	phySize := core.Alignphysize(int64(shard.Size))

	// allocate space
	pos, err := cd.allocSpace(phySize)
	if err != nil {
		return err
	}

	shard.Offset = pos

	headerbuf := make([]byte, core.GetShardHeaderSize())
	footerbuf := make([]byte, core.GetShardFooterSize())

	qoswAt := cd.qosWriterAt(ctx, cd.ef)

	// header
	err = shard.WriterHeader(headerbuf) // write header
	if err != nil {
		return err
	}

	start = time.Now()
	_, err = qoswAt.WriteAt(headerbuf, pos)
	span.AppendTrackLog("hdr.w", start, err)
	if err != nil {
		return err
	}

	pos += core.GetShardHeaderSize() // header + pos

	w = &bncomm.Writer{WriterAt: cd.ef, Offset: pos}
	twRaw := bncomm.NewTimeWriter(w)

	qosw := cd.qosWriter(ctx, twRaw)

	crc := crc32.NewIEEE()
	body := io.LimitReader(shard.Body, int64(shard.Size))
	body = io.TeeReader(body, crc)

	buffer = cd.pool.Get().([]byte)
	defer cd.pool.Put(buffer) // nolint: staticcheck

	tw := bncomm.NewTimeWriter(qosw)
	tr := bncomm.NewTimeReader(body)

	// write shard body
	encoder, err := crc32block.NewEncoder(buffer)
	if err != nil {
		return err
	}

	_, err = encoder.Encode(tr, int64(shard.Size), tw) // 写用户数据
	span.AppendTrackLogWithDuration("net.r", tr.Duration(), err)
	span.AppendTrackLogWithDuration("dat.w", twRaw.Duration(), err)
	span.AppendTrackLogWithDuration("dat.wai", tw.Duration()-twRaw.Duration(), err)

	if err != nil {
		if _, ok := err.(crc32block.ReaderError); ok {
			err = bloberr.ErrReaderError
		}
		return err
	}

	shard.Crc = crc.Sum32()

	// write footer
	err = shard.WriterFooter(footerbuf) // write footer
	if err != nil {
		return err
	}

	pos = w.Offset
	start = time.Now()

	_, err = qoswAt.WriteAt(footerbuf, pos) // write footer + pos
	span.AppendTrackLog("fo.w", start, err)
	if err != nil {
		return err
	}

	return nil
}

7. datafile 在 chunk file init 的时候初始化 wOff，
8. 在写入的时候， wOff 逻辑空间追加分配顺序，没有覆盖写的逻辑

const (
	_pageSize = 4 * 1024 // 4k
)


func (cd *datafile) init(meta *core.VuidMeta) (err error) {
	var sysstat syscall.Stat_t

	if sysstat, err = cd.ef.SysStat(); err != nil {
		return
	}

	chunkSize := sysstat.Size
	if chunkSize == 0 {
		// first time. auto format
		cd.initHeader(meta)
		if err = cd.writeMeta(); err != nil {
			return
		}
		cd.wOff = _chunkHeaderSize
	} else {
		if err = cd.parseMeta(); err != nil {
			return
		}
		cd.wOff = core.AlignSize(chunkSize, int64(_pageSize))
	}

	return
}


func (cd *datafile) allocSpace(fsize int64) (pos int64, err error) {
	cd.wLock.Lock()
	defer cd.wLock.Unlock()

	pos = cd.wOff

	cd.wOff += fsize
	cd.wOff = core.AlignSize(cd.wOff, _pageSize)

	return pos, nil
}

[jianwei1216]

===================================================================

疑问1：Chunk 大小有上限，这里的大小是指文件的逻辑空间，还是指实际占用的物理空间？

从代码来看，chunk file 的上限是指磁盘物理空间

1. ShardPut 的时候，会检查本次写入的 size 是否会超出 chunk 的上限，调用 HasEnoughSpace

//blobstore/blobnode/shard.go
/*
 *  method:         POST
 *  url:            /shard/put/diskid/{diskid}/vuid/{vuid}/bid/{bid}/size/{size}?iotype={iotype}
 *  request body:   bidData
 */
func (s *Service) ShardPut(c *rpc.Context) {
    args := new(bnapi.PutShardArgs)
    if err := c.ParseArgs(args); err != nil {
        c.RespondError(err)
        return
    }

    //这里判断当前的 chunk 是否还有空间可以写
    if !cs.HasEnoughSpace(args.Size) {
        span.Errorf("cs has no enougn space. args:%v, chunk info:%v, disk:%v",
                             args, cs.ChunkInfo(ctx), cs.Disk().Stats())
        c.RespondError(bloberr.ErrChunkNoSpace)
        return
    }

    shard := core.NewShardWriter(args.Bid, args.Vuid, uint32(args.Size), c.Request.Body)

    err = cs.Write(ctx, shard)
}

2. HasEnoughSpace 函数会判断 chunk 逻辑空间不超过 8T，物理空间不超过 16G * 2

const (
	DefaultDiskReservedSpaceB           = int64(60 << 30) // 60 GiB
	DefaultCompactReservedSpaceB        = int64(20 << 30) // 20 GiB
	DefaultChunkSize                    = int64(16 << 30) // 16 GiB
)


type ChunkAPI interface {
	HasEnoughSpace(needSize int64) bool
}

type chunk struct {
	// statics
	stats    ChunkStats
	fileInfo FileInfo
}

type FileInfo struct {
	Total uint64 `json:"total"` // ChunkSize (readonly) // 16 GiB
	Used  uint64 `json:"used"`  // user data size ( actual use size, phy size, live Update)
	Free  uint64 `json:"free"`  // ChunkSize - Used
	Size  uint64 `json:"size"`  // Chunk File Size ( file logic size)
}

const (
	DefaultChunkExpandRate  = 2
	DefaultMaxChunkFileSize = 8 * (1 << 40) // 8 TiB
)

func (cs *chunk) HasEnoughSpace(needSize int64) bool {
	size := atomic.LoadUint64(&cs.fileInfo.Size)  //文件逻辑空间的大小
	used := atomic.LoadUint64(&cs.fileInfo.Used)  // 实际物理磁盘空间的大小
	total := atomic.LoadUint64(&cs.fileInfo.Total) // 16GiB 物理磁盘空间的上限

    // chunk 文件的逻辑空间 + needSize 不能超过 8 TiB
	if size+uint64(needSize) >= DefaultMaxChunkFileSize { // 8 TiB
		return false
	}

    // chunk 文件的实际物理磁盘空间 + needSize 不能超过 16 * 2 = 32 GiB
	if used+uint64(needSize) >= total*DefaultChunkExpandRate { // total * 2 = 16 * 2 = 32 GiB �
		return false
	}

	diskStats := cs.Disk().Stats()
    // DefaultDiskReservedSpaceB = int64(60 << 30) // 60 GiB
	reserved := cs.conf.DiskReservedSpaceB
    //DefaultCompactReservedSpaceB = int64(20 << 30) // 20 GiB
	compactReserved := cs.conf.CompactReservedSpaceB 

    // needSize + 60 - 20 = needSize + 40 GiB < diskStats.Free
	return needSize+reserved-compactReserved < diskStats.Free
}

===================================================================
创建 chunk 时，会指定 chunk total

//blobstore/blobnode/chunk.go

/*
 *  method:         POST
 *  url:            /chunk/create/diskid/{diskid}/vuid/{vuid}?chunksize={chunksize}
 *  request body:   json.Marshal(bnapi.ChunkCreateArgs)
 */
func (s *Service) ChunkCreate(c *rpc.Context) {
	args := new(bnapi.CreateChunkArgs)
	if err := c.ParseArgs(args); err != nil {
		c.RespondError(err)
		return
	}

	if args.ChunkSize == 0 {
		args.ChunkSize = core.DefaultChunkSize //DefaultChunkSize = int64(16 << 30) // 16 GiB
	}

	cs, err := ds.CreateChunk(ctx, args.Vuid, args.ChunkSize) // ChunkSize = 16 GiB
}

type Service struct {
	Disks         map[proto.DiskID]core.DiskAPI
}

type DiskAPI interface {
	CreateChunk(ctx context.Context, vuid proto.Vuid, chunksize int64) (cs ChunkAPI, err error)
}

/*
 * 1. Create a new chunk
 * 2. bind it to vuid
 */
func (dsw *DiskStorageWrapper) CreateChunk(ctx context.Context, vuid proto.Vuid, chunksize int64) (
	cs core.ChunkAPI, err error) {
	span := trace.SpanFromContextSafe(ctx)

	ds := dsw.DiskStorage

	vm := core.VuidMeta{
		Version:   _chunkVer[0],
		Vuid:      vuid,
		DiskID:    ds.DiskID,
		ChunkId:   chunkId,
		ChunkSize: chunksize, //这里是 16G 上限
		Ctime:     nowtime,
		Mtime:     nowtime,
		Status:    bnapi.ChunkStatusNormal,
	}

	// create chunk storage
	cs, err = chunk.NewChunkStorage(ctx, ds.DataPath, vm, func(option *core.Option) {
		option.CreateDataIfMiss = true
		option.DB = ds.SuperBlock.db
		option.Conf = ds.Conf
		option.IoQos = ds.dataQos
		option.Disk = dsw
	})

//blobstore/blobnode/core/chunk/chunk.go
func NewChunkStorage(ctx context.Context, dataPath string, vm core.VuidMeta, opts ...core.OptionFunc) (
	cs *Chunk, err error,
) {
	c, err := newChunkStorage(ctx, dataPath, vm, opts...)
	if err != nil {
		return nil, err
	}

	cs = &Chunk{chunk: c}

	// It will be automatically recycled when gc
	runtime.SetFinalizer(cs, func(wapper *Chunk) {
		wapper.Close(context.Background())
	})

	return cs, nil
}

//blobstore/blobnode/core/chunk/chunk.go
func newChunkStorage(ctx context.Context, dataPath string, vm core.VuidMeta, opts ...core.OptionFunc) (
	cs *chunk, err error,
) {
	span := trace.SpanFromContextSafe(ctx)

	// chunk data
	chunkFile := filepath.Join(dataPath, vm.ChunkId.String())

	opt := core.Option{}
	for _, fn := range opts {
		fn(&opt)
	}

	// new chunkData fd
	cd, err := storage.NewChunkData(ctx, vm, chunkFile, opt.Conf, opt.CreateDataIfMiss, opt.IoQos)
	if err != nil {
		span.Errorf("Failed new chunk data. dp:%s, err:%v", dataPath, err)
		return nil, err
	}

	// create meta fd
	cm, err := storage.NewChunkMeta(ctx, opt.Conf, vm, opt.DB)
	if err != nil {
		span.Errorf("Failed new chunk meta. vm:%v, err:%v", vm, err)
		return nil, err
	}

	cs = &chunk{
		version:        vm.Version,
		vuid:           vm.Vuid,
		diskID:         vm.DiskID,
		disk:           opt.Disk,
		conf:           opt.Conf,
		status:         vm.Status,
		compacting:     vm.Compacting,
		bidlimiter:     keycount.NewBlockingKeyCountLimit(1),
		consistent:     core.NewConsistencyController(),
		lastModifyTime: vm.Mtime,
	}

	// init compact task
	cs.resetCompactTask()

	// init stg
	stg := storage.NewStorage(cm, cd)
	// enhence stg, with inline feat
	stg = storage.NewTinyFileStg(stg, opt.Conf.TinyFileThresholdB)

	cs.setStg(stg)

	cs.fileInfo.Total = uint64(vm.ChunkSize) // total 16G  这里指定 file total 为 16G
	err = cs.refreshFstat(ctx)
	if err != nil {
		span.Errorf("Failed chunk storage init, err:%v", err)
		return nil, err
	}

	return cs, err
}

[Cloudstriff]

1. 你上面的理解没有问题，blob是append的，chunk也是append的，但是volume并不是append的，volume是一个数据存储容器，只有volume有空间，就可以持续写入。而写入的时候到物理chunk上是一直append往后追加的。
2. 你的问题点在于chunk写满删除后怎么继续写入，实现上当chunk出现blob删除后，chunk物理空间大小会下降，此时chunk可以继续往后追加写入，即从逻辑上看chunk的大小最终会超过16GB，但是物理上会控制在16GB以内。
3. 当持续性的append写入和删除导致chunk的逻辑大小大于阈值时（这个阈值默值为 8T），会触发chunk级别的compaction，创建新的chunk file，将原先的chunk中的有效数据拷贝过去（全量+增量复制），避免单chunk punch hole出现严重的碎片化

[Cloudstriff]

@jianwei1216

[jianwei1216]

1. 当 blobstore volume 逻辑空间超过8T（实际物理空间可能只有1G）后，也会变为只读吗？
2. 如果变为只读的话，是不是需要一个volume下所有的 n+m 个 chunks 都完成 compact 后，才能够继续对外提供写能力？
3. 比较关注的是，chunk compact 是 blobnode粒度的（节点各自为政），还是 blobstore volume 粒度的（volume 下所有chunk 统一compact）