568f1fe5b1
When writing metadata logs to /topics/.system/log, the filer was not respecting the disk type configuration from path-specific rules (fs.configure). This caused volume assignment failures when volume servers used a specific disk type (e.g., "ssd") because the assign request defaulted to empty disk type. The fix adds DiskType to the VolumeAssignRequest in the filer's metadata log write path, ensuring that path-specific disk type configurations are properly honored for internal system writes. Fixes errors like: "metadata log write failed /topics/.system/log/...: AssignVolume: failed to find writable volumes for collection" Signed-off-by: Charles Darke <s.cduk@toodevious.com> Co-authored-by: Charles Darke <s.cduk@toodevious.com>
94 lines
2.6 KiB
Go
94 lines
2.6 KiB
Go
package filer
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/operation"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
)
|
|
|
|
func (f *Filer) appendToFile(targetFile string, data []byte) error {
|
|
|
|
assignResult, uploadResult, err2 := f.assignAndUpload(targetFile, data)
|
|
if err2 != nil {
|
|
return err2
|
|
}
|
|
|
|
// find out existing entry
|
|
fullpath := util.FullPath(targetFile)
|
|
entry, err := f.FindEntry(context.Background(), fullpath)
|
|
var offset int64 = 0
|
|
if err == filer_pb.ErrNotFound {
|
|
entry = &Entry{
|
|
FullPath: fullpath,
|
|
Attr: Attr{
|
|
Crtime: time.Now(),
|
|
Mtime: time.Now(),
|
|
Mode: os.FileMode(0644),
|
|
Uid: OS_UID,
|
|
Gid: OS_GID,
|
|
},
|
|
}
|
|
} else if err != nil {
|
|
return fmt.Errorf("find %s: %v", fullpath, err)
|
|
} else {
|
|
offset = int64(TotalSize(entry.GetChunks()))
|
|
}
|
|
|
|
// append to existing chunks
|
|
entry.Chunks = append(entry.GetChunks(), uploadResult.ToPbFileChunk(assignResult.Fid, offset, time.Now().UnixNano()))
|
|
|
|
// update the entry
|
|
err = f.CreateEntry(context.Background(), entry, false, false, nil, false, f.MaxFilenameLength)
|
|
|
|
return err
|
|
}
|
|
|
|
func (f *Filer) assignAndUpload(targetFile string, data []byte) (*operation.AssignResult, *operation.UploadResult, error) {
|
|
// assign a volume location
|
|
rule := f.FilerConf.MatchStorageRule(targetFile)
|
|
assignRequest := &operation.VolumeAssignRequest{
|
|
Count: 1,
|
|
Collection: util.Nvl(f.metaLogCollection, rule.Collection),
|
|
Replication: util.Nvl(f.metaLogReplication, rule.Replication),
|
|
DiskType: rule.DiskType,
|
|
WritableVolumeCount: rule.VolumeGrowthCount,
|
|
}
|
|
|
|
assignResult, err := operation.Assign(context.Background(), f.GetMaster, f.GrpcDialOption, assignRequest)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("AssignVolume: %w", err)
|
|
}
|
|
if assignResult.Error != "" {
|
|
return nil, nil, fmt.Errorf("AssignVolume error: %v", assignResult.Error)
|
|
}
|
|
|
|
// upload data
|
|
targetUrl := "http://" + assignResult.Url + "/" + assignResult.Fid
|
|
uploadOption := &operation.UploadOption{
|
|
UploadUrl: targetUrl,
|
|
Filename: "",
|
|
Cipher: f.Cipher,
|
|
IsInputCompressed: false,
|
|
MimeType: "",
|
|
PairMap: nil,
|
|
Jwt: assignResult.Auth,
|
|
}
|
|
|
|
uploader, err := operation.NewUploader()
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("upload data %s: %v", targetUrl, err)
|
|
}
|
|
|
|
uploadResult, err := uploader.UploadData(context.Background(), data, uploadOption)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("upload data %s: %v", targetUrl, err)
|
|
}
|
|
// println("uploaded to", targetUrl)
|
|
return assignResult, uploadResult, nil
|
|
}
|