III. Utility

Bit Rot Algorithm

通过类型 BitrotAlgorithm 来实现不同 Hash 算法的统一使用。例如在 minio 服务启动时，要执行 Hash 算法检查，如下所示

func bitrotSelfTest() {
    var checksums = map[BitrotAlgorithm]string{
        SHA256:          "a7677ff19e0182e4d52e3a3db727804abc82a5818749336369552e54b838b004",
        BLAKE2b512:      "e519b7d84b1c3c917985f544773a35cf265dcab10948be3550320d156bab612124a5ae2ae5a8c73c0eea360f68b0e28136f26e858756dbfe7375a7389f26c669",
        HighwayHash256:  "39c0407ed3f01b18d22c85db4aeff11e060ca5f43131b0126731ca197cd42313",
        HighwayHash256S: "39c0407ed3f01b18d22c85db4aeff11e060ca5f43131b0126731ca197cd42313",
    }
    for algorithm := range bitrotAlgorithms {
        if !algorithm.Available() {
            continue
        }
        checksum, err := hex.DecodeString(checksums[algorithm])
        if err != nil {
            logger.Fatal(errSelfTestFailure, fmt.Sprintf("bitrot: failed to decode %v checksum %s for selftest: %v", algorithm, checksums[algorithm], err))
        }
        var (
            hash = algorithm.New()
            msg  = make([]byte, 0, hash.Size()*hash.BlockSize())
            sum  = make([]byte, 0, hash.Size())
        )
        for i := 0; i < hash.Size()*hash.BlockSize(); i += hash.Size() {
            hash.Write(msg)
            sum = hash.Sum(sum[:0])
            msg = append(msg, sum...)
            hash.Reset()
        }
        if !bytes.Equal(sum, checksum) {
            logger.Fatal(errSelfTestFailure, fmt.Sprintf("bitrot: %v selftest checksum mismatch: got %x - want %x", algorithm, sum, checksum))
        }
    }
}

核心是通过新类型定义，统一 Hash 算法使用的规范，BitrotAlgorithm 定义如下

type BitrotAlgorithm uint
const (
    // SHA256 represents the SHA-256 hash function
    SHA256 BitrotAlgorithm = 1 + iota
    // HighwayHash256 represents the HighwayHash-256 hash function
    HighwayHash256
    // HighwayHash256S represents the Streaming HighwayHash-256 hash function
    HighwayHash256S
    // BLAKE2b512 represents the BLAKE2b-512 hash function
    BLAKE2b512
)

其 New 方法如下所示

func (a BitrotAlgorithm) New() hash.Hash {
    switch a {
    case SHA256:
        return sha256.New()
    case BLAKE2b512:
        b2, _ := blake2b.New512(nil) // New512 never returns an error if the key is nil
        return b2
    case HighwayHash256:
        hh, _ := highwayhash.New(magicHighwayHash256Key) // New will never return error since key is 256 bit
        return hh
    case HighwayHash256S:
        hh, _ := highwayhash.New(magicHighwayHash256Key) // New will never return error since key is 256 bit
        return hh
    default:
        logger.CriticalIf(GlobalContext, errors.New("Unsupported bitrot algorithm"))
        return nil
    }
}

Erasure Code

Minio 使用 Reed Solomon 纠错码来编码数据，算法实现部分使用 klauspost/reedsolomon。Minio 中封装如下

type Erasure struct {
    encoder                  func() reedsolomon.Encoder
    dataBlocks, parityBlocks int
    blockSize                int64
}
func NewErasure(ctx context.Context, dataBlocks, parityBlocks int, blockSize int64) (e Erasure, err error) {
    // Check the parameters for sanity now.
    if dataBlocks <= 0 || parityBlocks <= 0 {
        return e, reedsolomon.ErrInvShardNum
    }
    if dataBlocks+parityBlocks > 256 {
        return e, reedsolomon.ErrMaxShardNum
    }
    e = Erasure{
        dataBlocks:   dataBlocks,
        parityBlocks: parityBlocks,
        blockSize:    blockSize,
    }
    // Encoder when needed.
    var enc reedsolomon.Encoder
    var once sync.Once
    e.encoder = func() reedsolomon.Encoder {
        once.Do(func() {
            e, err := reedsolomon.New(dataBlocks, parityBlocks, reedsolomon.WithAutoGoroutines(int(e.ShardSize())))
            if err != nil {
                // Error conditions should be checked above.
                panic(err)
            }
            enc = e
        })
        return enc
    }
    return
}

编解码部分如下所示

// EncodeData encodes the given data and returns the erasure-coded data.
// It returns an error if the erasure coding failed.
func (e *Erasure) EncodeData(ctx context.Context, data []byte) ([][]byte, error) {
    if len(data) == 0 {
        return make([][]byte, e.dataBlocks+e.parityBlocks), nil
    }
    encoded, err := e.encoder().Split(data)
    if err != nil {
        logger.LogIf(ctx, err)
        return nil, err
    }
    if err = e.encoder().Encode(encoded); err != nil {
        logger.LogIf(ctx, err)
        return nil, err
    }
    return encoded, nil
}
// DecodeDataBlocks decodes the given erasure-coded data.
// It only decodes the data blocks but does not verify them.
// It returns an error if the decoding failed.
func (e *Erasure) DecodeDataBlocks(data [][]byte) error {
    var isZero = 0
    for _, b := range data[:] {
        if len(b) == 0 {
            isZero++
            break
        }
    }
    if isZero == 0 || isZero == len(data) {
        // If all are zero, payload is 0 bytes.
        return nil
    }
    return e.encoder().ReconstructData(data)
}
func (e *Erasure) DecodeDataAndParityBlocks(ctx context.Context, data [][]byte) error {
    if err := e.encoder().Reconstruct(data); err != nil {
        logger.LogIf(ctx, err)
        return err
    }
    return nil
}

Certificate Pool

图 1: x509 Certificate Pool AddCert 示意图

CertPool 来自于 Go 标准库，用于管理证书，在此处简要记录，方便后续理解。AddCert 可以清楚的看到 Certificate 及 CertPool 的关联，在 CertPool 的两个 map 中，value 域存储的是新增证书在切片中的索引值。