func (s *DiskStorage) SaveRaftEntries(entries []raftpb.Entry) error {
	if len(entries) == 0 {
		return nil
	}
	beginSaveTime := time.Now()
	startIndex := entries[0].Index
	records, recordsSize := entriesToRecords(entries)
	if SegmentSize < int64(recordsSize) {
		log.Log.Fatalf("recordsSize[%d] is large than SegmentSize[%d]", recordsSize, SegmentSize)
	}
	writeSegment, isNew, err := s.getWriteSegment(startIndex, recordsSize)
	if err != nil {
		log.Log.Errorf("getWriteSegment error,err:%s,index:%d,size:%d", err, entries[0].Index, recordsSize)
		return err
	}
	err = writeSegment.appendRecords(records, startIndex)
	if err != nil {
		log.Log.Errorf("appendRaftEntries,error,err:%s,len(entries):%d", err, len(entries))
		return err
	}
	if isNew {
		s.Mu.Lock()
		s.Segments = append(s.Segments, writeSegment)
		s.Mu.Unlock()
	}
	writeTps.Mark(int64(len(records)))
	writeThroughput.Mark(int64(recordsSize))
	writeLatency.Update(int64(time.Now().Sub(beginSaveTime) / time.Millisecond))
	return nil
}
SaveRaftEntries save raft entries in storage
package storage
import (
	"context"
	"fmt"
	"math"
	"os"
	"sync"
	"time"
	"github.com/flike/kingbus/utils"
	"hash/crc32"
	etcdraft "github.com/coreos/etcd/raft"
	"github.com/coreos/etcd/raft/raftpb"
	"github.com/flike/kingbus/log"
	"github.com/flike/kingbus/storage/storagepb"
	"go.uber.org/atomic"
)
var (
	noLimit uint64 = math.MaxUint64
	SegmentSize     int64 = 1024 * 1024 * 1024  
	commonSliceSize       = 200
)
const (
	FileMode = 0600
	DirMode = 0700
)
type DiskStorage struct {
	Dir string
	Mu       sync.RWMutex
	Segments []*Segment
	ReserveSegmentCount int
	purgeStarted        *atomic.Bool
	purgeCtx            context.Context
	purgeCancel         context.CancelFunc
	purgeQuitC          chan struct{}
	MetaStorage
}
func NewDiskStorage(dir string, reserveSizeInGB int) (*DiskStorage, error) {
	var err error
	if len(dir) == 0 {
		return nil, ErrArgsIllegal
	}
	s := new(DiskStorage)
	s.Dir = dir
	s.ReserveSegmentCount = reserveSizeInGB * 1024 * 1024 * 1024 / int(SegmentSize)
	if utils.DirExist(s.Dir) == false {
		err = os.MkdirAll(s.Dir, DirMode)
		if err != nil {
			return nil, err
		}
	}
	s.MetaStorage, err = NewMetaStore(s.Dir)
	if err != nil {
		log.Log.Errorf("NewMetaStore error,err:%s,dir:%s", err, dir)
		return nil, err
	}
	s.initSegments()
	s.purgeStarted = atomic.NewBool(false)
	s.setRecover(true)
	go s.StartPurgeLog()
	return s, nil
}
func (s *DiskStorage) initSegments() {
	if utils.ExistLog(s.Dir) == false {
		s.Segments = make([]*Segment, 0, 100)
		return
	}
	needRecover := s.getRecover()
	log.Log.Debugf("initSegments:needRecover is:%v", needRecover)
	s.Segments = openSegments(s.Dir, needRecover)
	return
}
func (s *DiskStorage) Entries(lo, hi, maxSize uint64) ([]raftpb.Entry, error) {
	last := hi - 1
	beginReadTime := time.Now()
	if last < lo {
		return nil, ErrArgsIllegal
	}
	err := s.checkRaftIndex(lo)
	if err != nil {
		log.Log.Errorf("checkRaftIndex error,err:%s,lo:%d,hi:%d", err, lo, hi)
		return nil, err
	}
	err = s.checkRaftIndex(last)
	if err != nil {
		log.Log.Errorf("checkRaftIndex error,err:%s,lo:%d,hi:%d", err, lo, hi)
		return nil, err
	}
	segments, err := s.getReadSegments(lo, last)
	if err != nil {
		log.Log.Errorf("checkRaftIndex error,err:%s,lo:%d,hi:%d", err, lo, hi)
		return nil, err
	}
	entries, err := s.readSegments(segments, lo, last, maxSize)
	if err != nil {
		log.Log.Errorf("checkRaftIndex error,err:%s,lo:%d,hi:%d", err, lo, hi)
		return nil, err
	}
	entries = s.cutRaftEntries(entries, maxSize)
	readLatency.Update(int64(time.Now().Sub(beginReadTime) / time.Millisecond))
	return entries, nil
}
func (s *DiskStorage) cutRaftEntries(entries []raftpb.Entry, maxSize uint64) []raftpb.Entry {
	if len(entries) == 0 {
		return entries
	}
	size := len(entries[0].Data)
	var limit int
	for limit = 1; limit < len(entries); limit++ {
		size += len(entries[limit].Data)
		if uint64(size) > maxSize {
			break
		}
	}
	for i := 0; i < len(entries)-1; i++ {
		if entries[i].Index+1 != entries[i+1].Index {
			for j := 0; j < len(entries); j++ {
				log.Log.Infof("j:%d, raftlog[j].Term:%d, raftlog[j].Index:%d,"+
					"raftlog[j].Type:%d, raftlog[j].Data:%v", j, entries[j].Term, entries[j].Index,
					entries[j].Type, entries[j].Data)
			}
			log.Log.Fatalf("cutRaftEntries:read raft log from storage is not continuous,i:%d", i)
		}
	}
	readTps.Mark(int64(len(entries)))
	readThroughput.Mark(int64(size))
	return entries[:limit]
}
func (s *DiskStorage) Term(i uint64) (uint64, error) {
	if i == 0 {
		return 0, nil
	}
	err := s.checkRaftIndex(i)
	if err != nil {
		log.Log.Errorf("checkRaftIndex error,err:%s,i:%d", err, i)
		return 0, err
	}
	entries, err := s.Entries(i, i+1, noLimit)
	if err != nil {
		return 0, err
	}
	if len(entries) == 0 {
		log.Log.Errorf("get term by index %d occur ErrUnavailable", i)
		return 0, etcdraft.ErrUnavailable
	}
	return entries[0].Term, nil
}
func (s *DiskStorage) LastIndex() (uint64, error) {
	s.Mu.RLock()
	defer s.Mu.RUnlock()
	if len(s.Segments) == 0 {
		return 0, nil
	}
	return s.Segments[len(s.Segments)-1].getLastIndex(), nil
}
func (s *DiskStorage) FirstIndex() (uint64, error) {
	s.Mu.RLock()
	defer s.Mu.RUnlock()
	if len(s.Segments) == 0 {
		return 1, nil
	}
	return s.Segments[0].getFirstIndex(), nil
}
func (s *DiskStorage) Snapshot() (raftpb.Snapshot, error) {
	return raftpb.Snapshot{}, etcdraft.ErrSnapshotTemporarilyUnavailable
}
func (s *DiskStorage) TruncateSuffix(raftIndex uint64) error {
	var (
		truncateSegment *Segment
		segments        []*Segment
		err             error
	)
	s.Mu.Lock()
	segmentIndex := -1
	for i := len(s.Segments) - 1; 0 <= i; i-- {
		l := s.Segments[i].getFirstIndex()
		h := s.Segments[i].getLastIndex()
		if l <= raftIndex && raftIndex <= h {
			segmentIndex = i
			truncateSegment = s.Segments[i]
			if i < len(s.Segments)-1 {
				segments = s.Segments[i+1:]
				s.Segments = s.Segments[:i+1]
			}
			break
		}
	}
	s.Mu.Unlock()
	if segmentIndex == -1 {
		return ErrOutOfBound
	}
	if len(segments) != 0 {
		err = s.purgeSegments(segments, false)
		if err != nil {
			log.Log.Errorf("purgeSegments error,err:%s,len(segments):%v", err, len(segments))
			return err
		}
	}
	if truncateSegment != nil {
		err = truncateSegment.truncateSuffix(raftIndex)
		if err != nil {
			log.Log.Errorf("truncateSuffix error,err:%s,segmentPath:%s,raftIndex:%d",
				err, truncateSegment.LogFile.filePath, raftIndex)
			return err
		}
	}
	return nil
}
func (s *DiskStorage) purgeSegments(segments []*Segment, delay bool) error {
	if delay {
		<-time.After(time.Second * 4)
	}
	firstIndex, err := s.FirstIndex()
	if err != nil {
		return err
	}
	err = s.MetaStorage.UpdatePugedGtidset(firstIndex)
	if err != nil {
		return err
	}
	for i := 0; i < len(segments); i++ {
		err := segments[i].remove()
		if err != nil {
			log.Log.Errorf("segment remove error,err:%s,segmentPath:%s", err, segments[i].LogFile.filePath)
			return err
		}
	}
	return nil
}