mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-20 13:30:46 +02:00
format: add adapter registry mapping file structure to chunk extents
A format adapter reduces one container format to three things the core understands: extent sizes, an alignment quantum, and an opaque payload. Capabilities beyond identity (Indexer, SidecarIndexer, Viewer) are discovered by type assertion. The layout persists in one compact extended attribute keyed by extent sizes rather than chunk ids, so it survives chunk manifest folding, and the Cutter turns it into upload chunk boundaries clamped by maxMB and the align quantum. The formattest kit holds every adapter to no-panic parsing of truncated input.
This commit is contained in:
Chris Lu
@@ -0,0 +1,83 @@
|
||||
// Package format maps the internal structure of container file formats onto
|
||||
// storage chunk boundaries.
|
||||
//
|
||||
// An adapter translates one format into three things the core understands: a
|
||||
// list of extent sizes, an alignment quantum, and an opaque payload. Adapters
|
||||
// never see chunks, file ids, or authorization; the core never learns what an
|
||||
// MPEG-TS packet or a parquet row group is.
|
||||
package format
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net/url"
|
||||
)
|
||||
|
||||
// LayoutKey is the filer Extended key holding the encoded Layout. The
|
||||
// x-seaweedfs- prefix keeps it out of HTTP response headers.
|
||||
const LayoutKey = "x-seaweedfs-format-layout"
|
||||
|
||||
// ErrNoSuchView reports that a view request addresses nothing servable; the
|
||||
// server answers 404.
|
||||
var ErrNoSuchView = errors.New("no such view")
|
||||
|
||||
// Layout describes how a file's structure maps to byte extents.
|
||||
type Layout struct {
|
||||
Format string // adapter name
|
||||
ExtentSizes []int64 // extent lengths in file order; they sum to the file size
|
||||
Align int64 // quantum for cutting inside an oversized extent; 1 cuts anywhere
|
||||
Payload []byte // adapter-owned metadata, opaque to the core
|
||||
}
|
||||
|
||||
// Hint carries the cheap identification signals available to Sniff.
|
||||
type Hint struct {
|
||||
Name string
|
||||
ContentType string
|
||||
Size int64
|
||||
Head []byte
|
||||
Tail []byte
|
||||
}
|
||||
|
||||
// Format is the mandatory adapter identity. Capabilities beyond it are
|
||||
// discovered by type assertion.
|
||||
type Format interface {
|
||||
Name() string
|
||||
Sniff(h Hint) bool
|
||||
}
|
||||
|
||||
// Indexer derives a Layout from the complete stored bytes.
|
||||
type Indexer interface {
|
||||
Index(ctx context.Context, r io.ReaderAt, size int64) (*Layout, error)
|
||||
}
|
||||
|
||||
// SidecarIndexer derives a Layout from an external index document supplied at
|
||||
// ingest, before the media bytes arrive.
|
||||
type SidecarIndexer interface {
|
||||
IndexSidecar(sidecar []byte) (*Layout, error)
|
||||
}
|
||||
|
||||
// Object is everything a Viewer may know about the file it serves.
|
||||
type Object struct {
|
||||
Name string
|
||||
Size int64
|
||||
Layout *Layout
|
||||
}
|
||||
|
||||
// ViewRequest carries the request parameters of a ?view= request.
|
||||
type ViewRequest struct {
|
||||
Query url.Values
|
||||
}
|
||||
|
||||
// ViewPlan tells the server what to serve. The server executes it on the
|
||||
// normal streaming path; adapters stay pure functions of request and layout.
|
||||
type ViewPlan struct {
|
||||
ContentType string
|
||||
Body []byte // rendered document; when nil, stream Extent instead
|
||||
Extent int
|
||||
}
|
||||
|
||||
// Viewer answers ?view= requests.
|
||||
type Viewer interface {
|
||||
View(req ViewRequest, obj Object) (*ViewPlan, error)
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
// Package formattest is the conformance kit format adapters must pass:
|
||||
// indexers parse attacker-controlled bytes inside a storage daemon, so they
|
||||
// must never panic and every layout they accept must validate.
|
||||
package formattest
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/format"
|
||||
)
|
||||
|
||||
// IndexTruncations feeds progressively truncated copies of a valid file to the
|
||||
// indexer. Any outcome is acceptable except a panic or an invalid layout.
|
||||
func IndexTruncations(t *testing.T, indexer format.Indexer, data []byte) {
|
||||
t.Helper()
|
||||
for i := 0; i <= 16; i++ {
|
||||
size := int64(len(data) * i / 16)
|
||||
layout, err := indexer.Index(context.Background(), bytes.NewReader(data[:size]), size)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if validateErr := layout.Validate(size); validateErr != nil {
|
||||
t.Fatalf("Index() at %d/%d bytes returned an invalid layout: %v", size, len(data), validateErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// SidecarTruncations does the same for sidecar index documents.
|
||||
func SidecarTruncations(t *testing.T, indexer format.SidecarIndexer, sidecar []byte) {
|
||||
t.Helper()
|
||||
for i := 0; i <= 16; i++ {
|
||||
layout, err := indexer.IndexSidecar(sidecar[:len(sidecar)*i/16])
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if validateErr := layout.Validate(-1); validateErr != nil {
|
||||
t.Fatalf("IndexSidecar() at %d/16 returned an invalid layout: %v", i, validateErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// EncodeRoundTrip checks that a layout survives the persistence codec.
|
||||
func EncodeRoundTrip(t *testing.T, layout *format.Layout) {
|
||||
t.Helper()
|
||||
encoded, err := layout.Encode()
|
||||
if err != nil {
|
||||
t.Fatalf("Encode() error = %v", err)
|
||||
}
|
||||
decoded, err := format.DecodeLayout(encoded)
|
||||
if err != nil {
|
||||
t.Fatalf("DecodeLayout() error = %v", err)
|
||||
}
|
||||
if decoded.Format != layout.Format || decoded.Align != layout.Align ||
|
||||
len(decoded.ExtentSizes) != len(layout.ExtentSizes) || !bytes.Equal(decoded.Payload, layout.Payload) {
|
||||
t.Fatalf("decoded layout %+v differs from %+v", decoded, layout)
|
||||
}
|
||||
for i := range layout.ExtentSizes {
|
||||
if decoded.ExtentSizes[i] != layout.ExtentSizes[i] {
|
||||
t.Fatalf("extent %d = %d, want %d", i, decoded.ExtentSizes[i], layout.ExtentSizes[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,191 @@
|
||||
package format
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"math"
|
||||
"sort"
|
||||
)
|
||||
|
||||
const (
|
||||
layoutVersion = 1
|
||||
|
||||
// MaxExtentCount bounds decoded layouts; it also bounds the chunk count a
|
||||
// layout can force on an entry.
|
||||
MaxExtentCount = 1 << 20
|
||||
// MaxPayloadBytes bounds the adapter payload carried in entry metadata.
|
||||
MaxPayloadBytes = 16 << 20
|
||||
)
|
||||
|
||||
// TotalSize returns the file size the layout describes.
|
||||
func (l *Layout) TotalSize() int64 {
|
||||
var total int64
|
||||
for _, size := range l.ExtentSizes {
|
||||
total += size
|
||||
}
|
||||
return total
|
||||
}
|
||||
|
||||
// ExtentRange returns the byte range of extent i.
|
||||
func (l *Layout) ExtentRange(i int) (offset, size int64, ok bool) {
|
||||
if i < 0 || i >= len(l.ExtentSizes) {
|
||||
return 0, 0, false
|
||||
}
|
||||
for _, extentSize := range l.ExtentSizes[:i] {
|
||||
offset += extentSize
|
||||
}
|
||||
return offset, l.ExtentSizes[i], true
|
||||
}
|
||||
|
||||
// Validate checks layout consistency. A negative fileSize skips the total size
|
||||
// check.
|
||||
func (l *Layout) Validate(fileSize int64) error {
|
||||
if l.Format == "" {
|
||||
return fmt.Errorf("layout has no format name")
|
||||
}
|
||||
if l.Align < 1 {
|
||||
return fmt.Errorf("layout align %d is invalid", l.Align)
|
||||
}
|
||||
if len(l.ExtentSizes) == 0 {
|
||||
return fmt.Errorf("layout has no extents")
|
||||
}
|
||||
if len(l.ExtentSizes) > MaxExtentCount {
|
||||
return fmt.Errorf("layout has too many extents: %d", len(l.ExtentSizes))
|
||||
}
|
||||
if len(l.Payload) > MaxPayloadBytes {
|
||||
return fmt.Errorf("layout payload is too large: %d bytes", len(l.Payload))
|
||||
}
|
||||
var total int64
|
||||
for i, size := range l.ExtentSizes {
|
||||
if size <= 0 {
|
||||
return fmt.Errorf("extent %d has invalid size %d", i, size)
|
||||
}
|
||||
if size > math.MaxInt64-total {
|
||||
return fmt.Errorf("extent %d overflows the file size", i)
|
||||
}
|
||||
total += size
|
||||
}
|
||||
if fileSize >= 0 && total != fileSize {
|
||||
return fmt.Errorf("layout describes %d bytes but the file has %d", total, fileSize)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Encode serializes the layout for the LayoutKey extended attribute.
|
||||
func (l *Layout) Encode() ([]byte, error) {
|
||||
if err := l.Validate(-1); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := []byte{layoutVersion}
|
||||
out = binary.AppendUvarint(out, uint64(len(l.Format)))
|
||||
out = append(out, l.Format...)
|
||||
out = binary.AppendUvarint(out, uint64(l.Align))
|
||||
out = binary.AppendUvarint(out, uint64(len(l.ExtentSizes)))
|
||||
for _, size := range l.ExtentSizes {
|
||||
out = binary.AppendUvarint(out, uint64(size))
|
||||
}
|
||||
out = binary.AppendUvarint(out, uint64(len(l.Payload)))
|
||||
out = append(out, l.Payload...)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// DecodeLayout parses an encoded layout and validates it.
|
||||
func DecodeLayout(data []byte) (*Layout, error) {
|
||||
reader := bytes.NewReader(data)
|
||||
version, err := reader.ReadByte()
|
||||
if err != nil || version != layoutVersion {
|
||||
return nil, fmt.Errorf("unsupported layout version")
|
||||
}
|
||||
name, err := readUvarintBytes(reader, 256)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read layout format: %w", err)
|
||||
}
|
||||
align, err := binary.ReadUvarint(reader)
|
||||
if err != nil || align > math.MaxInt64 {
|
||||
return nil, fmt.Errorf("read layout align: invalid")
|
||||
}
|
||||
count, err := binary.ReadUvarint(reader)
|
||||
if err != nil || count > MaxExtentCount {
|
||||
return nil, fmt.Errorf("read layout extent count: invalid")
|
||||
}
|
||||
sizes := make([]int64, count)
|
||||
for i := range sizes {
|
||||
size, err := binary.ReadUvarint(reader)
|
||||
if err != nil || size > math.MaxInt64 {
|
||||
return nil, fmt.Errorf("read extent %d size: invalid", i)
|
||||
}
|
||||
sizes[i] = int64(size)
|
||||
}
|
||||
payload, err := readUvarintBytes(reader, MaxPayloadBytes)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read layout payload: %w", err)
|
||||
}
|
||||
if reader.Len() != 0 {
|
||||
return nil, fmt.Errorf("layout has %d trailing bytes", reader.Len())
|
||||
}
|
||||
layout := &Layout{Format: string(name), ExtentSizes: sizes, Align: int64(align), Payload: payload}
|
||||
if err := layout.Validate(-1); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return layout, nil
|
||||
}
|
||||
|
||||
func readUvarintBytes(reader *bytes.Reader, limit uint64) ([]byte, error) {
|
||||
length, err := binary.ReadUvarint(reader)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if length > limit || length > uint64(reader.Len()) {
|
||||
return nil, fmt.Errorf("length %d is out of bounds", length)
|
||||
}
|
||||
if length == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
data := make([]byte, length)
|
||||
if _, err := reader.Read(data); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return data, nil
|
||||
}
|
||||
|
||||
// Cutter yields upload chunk boundaries: every extent boundary, plus
|
||||
// align-quantized cuts inside extents larger than maxChunkSize. A non-positive
|
||||
// maxChunkSize keeps each extent in one chunk.
|
||||
type Cutter struct {
|
||||
cuts []int64 // absolute end offset of every chunk, ascending
|
||||
}
|
||||
|
||||
func (l *Layout) Cutter(maxChunkSize int64) *Cutter {
|
||||
quantum := maxChunkSize
|
||||
if quantum > 0 && l.Align > 1 {
|
||||
quantum -= quantum % l.Align
|
||||
if quantum <= 0 {
|
||||
// An align larger than the chunk limit still cuts on whole atoms.
|
||||
quantum = l.Align
|
||||
}
|
||||
}
|
||||
var cuts []int64
|
||||
var offset int64
|
||||
for _, size := range l.ExtentSizes {
|
||||
end := offset + size
|
||||
if quantum > 0 {
|
||||
for next := offset + quantum; next < end; next += quantum {
|
||||
cuts = append(cuts, next)
|
||||
}
|
||||
}
|
||||
cuts = append(cuts, end)
|
||||
offset = end
|
||||
}
|
||||
return &Cutter{cuts: cuts}
|
||||
}
|
||||
|
||||
// NextChunkSize returns the size of the chunk starting at offset, or 0 past
|
||||
// the end. It satisfies the filer upload loop's ChunkBoundaries interface.
|
||||
func (c *Cutter) NextChunkSize(offset int64) int64 {
|
||||
i := sort.Search(len(c.cuts), func(i int) bool { return c.cuts[i] > offset })
|
||||
if i == len(c.cuts) {
|
||||
return 0
|
||||
}
|
||||
return c.cuts[i] - offset
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
package format
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestLayoutEncodeDecodeRoundTrip(t *testing.T) {
|
||||
layout := &Layout{
|
||||
Format: "hls-ts",
|
||||
ExtentSizes: []int64{188 * 3, 188 * 2, 188 * 7},
|
||||
Align: 188,
|
||||
Payload: []byte{1, 2, 3},
|
||||
}
|
||||
encoded, err := layout.Encode()
|
||||
if err != nil {
|
||||
t.Fatalf("Encode() error = %v", err)
|
||||
}
|
||||
decoded, err := DecodeLayout(encoded)
|
||||
if err != nil {
|
||||
t.Fatalf("DecodeLayout() error = %v", err)
|
||||
}
|
||||
if decoded.Format != layout.Format || decoded.Align != layout.Align {
|
||||
t.Fatalf("decoded = %+v, want %+v", decoded, layout)
|
||||
}
|
||||
if len(decoded.ExtentSizes) != len(layout.ExtentSizes) {
|
||||
t.Fatalf("extent count = %d, want %d", len(decoded.ExtentSizes), len(layout.ExtentSizes))
|
||||
}
|
||||
for i := range layout.ExtentSizes {
|
||||
if decoded.ExtentSizes[i] != layout.ExtentSizes[i] {
|
||||
t.Fatalf("extent %d = %d, want %d", i, decoded.ExtentSizes[i], layout.ExtentSizes[i])
|
||||
}
|
||||
}
|
||||
if string(decoded.Payload) != string(layout.Payload) {
|
||||
t.Fatalf("payload = %v, want %v", decoded.Payload, layout.Payload)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeLayoutRejectsCorruptInput(t *testing.T) {
|
||||
layout := &Layout{Format: "parquet", ExtentSizes: []int64{10, 20}, Align: 1}
|
||||
encoded, err := layout.Encode()
|
||||
if err != nil {
|
||||
t.Fatalf("Encode() error = %v", err)
|
||||
}
|
||||
for cut := 0; cut < len(encoded); cut++ {
|
||||
if _, err := DecodeLayout(encoded[:cut]); err == nil {
|
||||
t.Fatalf("DecodeLayout() accepted truncation at %d", cut)
|
||||
}
|
||||
}
|
||||
if _, err := DecodeLayout(append(append([]byte{}, encoded...), 0)); err == nil {
|
||||
t.Fatalf("DecodeLayout() accepted trailing bytes")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLayoutValidate(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
layout Layout
|
||||
fileSize int64
|
||||
wantErr string
|
||||
}{
|
||||
{"valid", Layout{Format: "x", ExtentSizes: []int64{5, 5}, Align: 1}, 10, ""},
|
||||
{"skip size check", Layout{Format: "x", ExtentSizes: []int64{5}, Align: 1}, -1, ""},
|
||||
{"wrong total", Layout{Format: "x", ExtentSizes: []int64{5, 5}, Align: 1}, 11, "but the file has"},
|
||||
{"zero extent", Layout{Format: "x", ExtentSizes: []int64{5, 0}, Align: 1}, -1, "invalid size"},
|
||||
{"no extents", Layout{Format: "x", Align: 1}, -1, "no extents"},
|
||||
{"bad align", Layout{Format: "x", ExtentSizes: []int64{5}, Align: 0}, -1, "align"},
|
||||
{"no name", Layout{ExtentSizes: []int64{5}, Align: 1}, -1, "format name"},
|
||||
}
|
||||
for _, test := range tests {
|
||||
err := test.layout.Validate(test.fileSize)
|
||||
if test.wantErr == "" {
|
||||
if err != nil {
|
||||
t.Fatalf("%s: Validate() error = %v", test.name, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err == nil || !strings.Contains(err.Error(), test.wantErr) {
|
||||
t.Fatalf("%s: Validate() error = %v, want %q", test.name, err, test.wantErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtentRange(t *testing.T) {
|
||||
layout := &Layout{Format: "x", ExtentSizes: []int64{10, 20, 30}, Align: 1}
|
||||
offset, size, ok := layout.ExtentRange(1)
|
||||
if !ok || offset != 10 || size != 20 {
|
||||
t.Fatalf("ExtentRange(1) = (%d, %d, %v), want (10, 20, true)", offset, size, ok)
|
||||
}
|
||||
if _, _, ok := layout.ExtentRange(3); ok {
|
||||
t.Fatalf("ExtentRange(3) accepted out-of-range index")
|
||||
}
|
||||
if _, _, ok := layout.ExtentRange(-1); ok {
|
||||
t.Fatalf("ExtentRange(-1) accepted negative index")
|
||||
}
|
||||
}
|
||||
|
||||
// collectChunks walks the cutter the way the upload loop does.
|
||||
func collectChunks(t *testing.T, cutter *Cutter) [][2]int64 {
|
||||
t.Helper()
|
||||
var chunks [][2]int64
|
||||
var offset int64
|
||||
for {
|
||||
size := cutter.NextChunkSize(offset)
|
||||
if size <= 0 {
|
||||
return chunks
|
||||
}
|
||||
chunks = append(chunks, [2]int64{offset, size})
|
||||
offset += size
|
||||
}
|
||||
}
|
||||
|
||||
func TestCutterKeepsExtentBoundaries(t *testing.T) {
|
||||
layout := &Layout{Format: "x", ExtentSizes: []int64{5, 4}, Align: 1}
|
||||
chunks := collectChunks(t, layout.Cutter(16))
|
||||
want := [][2]int64{{0, 5}, {5, 4}}
|
||||
if len(chunks) != len(want) {
|
||||
t.Fatalf("chunks = %v, want %v", chunks, want)
|
||||
}
|
||||
for i := range want {
|
||||
if chunks[i] != want[i] {
|
||||
t.Fatalf("chunk %d = %v, want %v", i, chunks[i], want[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCutterSplitsOversizedExtentsOnAlign(t *testing.T) {
|
||||
// maxChunkSize 5 with align 2 quantizes down to 4-byte interior cuts.
|
||||
layout := &Layout{Format: "x", ExtentSizes: []int64{10, 3}, Align: 2}
|
||||
chunks := collectChunks(t, layout.Cutter(5))
|
||||
want := [][2]int64{{0, 4}, {4, 4}, {8, 2}, {10, 3}}
|
||||
if len(chunks) != len(want) {
|
||||
t.Fatalf("chunks = %v, want %v", chunks, want)
|
||||
}
|
||||
for i := range want {
|
||||
if chunks[i] != want[i] {
|
||||
t.Fatalf("chunk %d = %v, want %v", i, chunks[i], want[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCutterAlignLargerThanChunkLimit(t *testing.T) {
|
||||
// Align above maxChunkSize still cuts on whole atoms.
|
||||
layout := &Layout{Format: "x", ExtentSizes: []int64{20}, Align: 8}
|
||||
chunks := collectChunks(t, layout.Cutter(5))
|
||||
want := [][2]int64{{0, 8}, {8, 8}, {16, 4}}
|
||||
for i := range want {
|
||||
if chunks[i] != want[i] {
|
||||
t.Fatalf("chunk %d = %v, want %v", i, chunks[i], want[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCutterUnlimitedKeepsOneChunkPerExtent(t *testing.T) {
|
||||
layout := &Layout{Format: "x", ExtentSizes: []int64{10, 3}, Align: 188}
|
||||
chunks := collectChunks(t, layout.Cutter(0))
|
||||
want := [][2]int64{{0, 10}, {10, 3}}
|
||||
if len(chunks) != len(want) {
|
||||
t.Fatalf("chunks = %v, want %v", chunks, want)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package format
|
||||
|
||||
// Adapters register from init(), so the map needs no locking.
|
||||
var formats = map[string]Format{}
|
||||
|
||||
func Register(f Format) {
|
||||
if _, ok := formats[f.Name()]; ok {
|
||||
panic("format: duplicate adapter " + f.Name())
|
||||
}
|
||||
formats[f.Name()] = f
|
||||
}
|
||||
|
||||
// ByName returns the registered adapter, or nil.
|
||||
func ByName(name string) Format {
|
||||
return formats[name]
|
||||
}
|
||||
Reference in New Issue
Block a user