scan_vlog.go

// Copyright 2019 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package main

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"hash/crc32"
	"io"
	"log"
	"os"
)

const (
	headerLength     int64  = 16 // 4 + 8 + 4 magic + length + checksum
	recordMagic      uint32 = 0x823a56e8
	fileFooterLength int64  = 4 + 8 // fileEndMagic + maxTS
)

var crcTable = crc32.MakeTable(crc32.Castagnoli)

type Record struct {
	magic    uint32
	length   uint64
	checksum uint32
	payload  []byte
}

func (r *Record) readHeader(reader io.Reader) error {
	err := binary.Read(reader, binary.LittleEndian, &r.magic)
	if err != nil {
		return err
	}
	err = binary.Read(reader, binary.LittleEndian, &r.length)
	if err != nil {
		return err
	}
	err = binary.Read(reader, binary.LittleEndian, &r.checksum)
	if err != nil {
		return err
	}

	return nil
}

func (r *Record) isValid() bool {
	return crc32.Checksum(r.payload, crcTable) == r.checksum
}

func (r *Record) recordLength() int64 {
	return headerLength + int64(len(r.payload))
}

func getReader(f *os.File, offset, fileSize int64) *bufio.Reader {
	return bufio.NewReader(io.NewSectionReader(f, offset, fileSize-offset))
}

func seekToNextRecord(reader *bufio.Reader) (bytes int, err error) {
	var buf []byte
	for {
		// Peek 4 bytes to see if it matches recordMagic
		buf, err = reader.Peek(4)
		if err != nil {
			bytes += len(buf)
			return
		}

		magic := binary.LittleEndian.Uint32(buf)
		if magic == recordMagic {
			return
		}

		reader.Discard(1)
		bytes++
	}
}

func readRecord(reader io.Reader) (record *Record, err error) {
	record = new(Record)
	err = record.readHeader(reader)
	if err != nil {
		return nil, err
	}

	if record.magic != recordMagic {
		return nil, errors.New("invalid magic")
	}

	// read directly to record.payload if record.length is less than some value
	// note the data may be corrupt and record.length is not the true value
	if record.length < (4 << 30) {
		record.payload = make([]byte, record.length)
		_, err = io.ReadFull(reader, record.payload)
		if err != nil {
			return nil, err
		}
	} else {
		buf := new(bytes.Buffer)
		_, err = io.CopyN(buf, reader, int64(record.length))
		if err != nil {
			return
		}
		record.payload = buf.Bytes()
	}

	if !record.isValid() {
		return nil, errors.New("checksum mismatch")
	}

	return
}

func scanVLog(f *os.File) error {
	info, err := f.Stat()
	if err != nil {
		return err
	}

	size := info.Size()
	size -= fileFooterLength

	offset := int64(0)
	var reader = getReader(f, offset, size)
	for offset < size {
		r, err := readRecord(reader)
		if err != nil {
			offset = offset + 1
			reader = getReader(f, offset, size)
			bytes, seekErr := seekToNextRecord(reader)
			if seekErr == nil {
				log.Printf("Corruption detected, starting at %d, length %d", offset, bytes)
				offset += int64(bytes)
				continue
			}

			// reach file end
			if seekErr == io.EOF {
				log.Printf("Corruption detected, starting at %d till the end", offset)
				return nil
			}

			return seekErr
		}
		recordLen := r.recordLength()
		log.Printf("Read OK at %d", offset)
		offset += recordLen
	}

	return nil
}

func main() {
	fileName := os.Args[1]
	fmt.Printf("Scanning %s\n", fileName)
	f, err := os.Open(fileName)
	if err != nil {
		log.Fatalf("Failed to open file %s: %v", fileName, err)
	}
	scanVLog(f)
}