cache/chunk/chunk.go

Rama main · réplica de lectura del repositorio original.

package chunk

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"os"
	"unsafe"

	lz4 "github.com/DataDog/golz4"
	"github.com/coroot/coroot/model"
	"github.com/coroot/coroot/timeseries"
)

const (
	V1 uint8 = 1
	V2 uint8 = 2
	V3 uint8 = 3
	V4 uint8 = 4

	Size = timeseries.Minute * 10
)

type Meta struct {
	Path        string
	From        timeseries.Time
	PointsCount uint32
	Step        timeseries.Duration
	Finalized   bool
	Created     timeseries.Time
}

func (m *Meta) To() timeseries.Time {
	return m.From.Add(timeseries.Duration(m.PointsCount-1) * m.Step)
}

func (m *Meta) Jitter() timeseries.Duration {
	return m.From.Sub(m.From.Truncate(Size))
}

type metricMeta struct {
	Hash       uint64
	MetaOffset uint32
	MetaSize   uint32
}

const metricMetaSize = 16

var ErrCorrupted = errors.New("corrupted chunk")

type header struct {
	Version     uint8
	From        timeseries.Time
	PointsCount uint32
	Step        timeseries.Duration
	Finalized   bool

	DataSizeOrMetricsCount uint32
}

func ReadMeta(path string) (*Meta, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	defer f.Close()

	stat, err := f.Stat()
	if err != nil {
		return nil, err
	}
	h := header{}
	if err = binary.Read(f, binary.LittleEndian, &h); err != nil {
		return nil, err
	}
	if h.Version != V3 && h.Version != V4 {
		return nil, fmt.Errorf("%w: unknown version %d", ErrCorrupted, h.Version)
	}
	if h.PointsCount == 0 || h.Step <= 0 {
		return nil, fmt.Errorf("%w: invalid header: points_count=%d, step=%d", ErrCorrupted, h.PointsCount, h.Step)
	}
	return &Meta{
		Path:        path,
		From:        h.From,
		PointsCount: h.PointsCount,
		Step:        h.Step,
		Finalized:   h.Finalized,
		Created:     timeseries.TimeFromStandard(stat.ModTime()),
	}, err
}

func Read(path string, from timeseries.Time, pointsCount int, step timeseries.Duration, dest map[uint64]*model.MetricValues, fillFunc timeseries.FillFunc) error {
	f, err := os.Open(path)
	if err != nil {
		return err
	}
	defer f.Close()

	reader := bufio.NewReader(f)
	h := header{}
	if err = binary.Read(reader, binary.LittleEndian, &h); err != nil {
		return err
	}
	switch h.Version {
	case V3:
		return readV3(reader, &h, from, pointsCount, step, dest, fillFunc)
	case V4:
		return readV4(reader, &h, from, pointsCount, step, dest, fillFunc)
	default:
		return fmt.Errorf("%w: unknown version %d", ErrCorrupted, h.Version)
	}
}

func readV3(reader io.Reader, header *header, from timeseries.Time, pointsCount int, step timeseries.Duration, dest map[uint64]*model.MetricValues, fillFunc timeseries.FillFunc) error {
	r := lz4.NewDecompressReader(reader)
	defer r.Close()
	buf := make([]byte, metricMetaSize+4*header.PointsCount)
	var labelsToRead []*metricMeta
	var maxLabelSize uint32
	var err error
	for range header.DataSizeOrMetricsCount {
		if _, err = io.ReadFull(r, buf); err != nil {
			return err
		}
		m := metricMeta{
			Hash:       binary.LittleEndian.Uint64(buf),
			MetaOffset: binary.LittleEndian.Uint32(buf[8:]),
			MetaSize:   binary.LittleEndian.Uint32(buf[12:]),
		}
		mv, ok := dest[m.Hash]
		if mv == nil {
			mv = &model.MetricValues{
				Values: timeseries.New(from, pointsCount, step),
			}
			labelsToRead = append(labelsToRead, &m)
			if m.MetaSize > maxLabelSize {
				maxLabelSize = m.MetaSize
			}
		}
		if !fillFunc(mv.Values, header.From, header.Step, asFloats32(buf[metricMetaSize:])) && !ok {
			continue
		}
		dest[m.Hash] = mv
	}
	if len(labelsToRead) > 0 {
		buf = make([]byte, maxLabelSize)
		offset := uint32(0)
		for _, m := range labelsToRead {
			mv, ok := dest[m.Hash]
			if !ok {
				continue
			}
			toSkip := m.MetaOffset - offset
			if toSkip > 0 {
				if _, err := io.CopyN(io.Discard, r, int64(toSkip)); err != nil {
					return err
				}
			}
			if _, err = io.ReadFull(r, buf[:m.MetaSize]); err != nil {
				return err
			}
			offset = m.MetaOffset + m.MetaSize
			mv.LabelsHash = m.Hash
			mv.Labels = make(model.Labels)
			readLabels(buf[:m.MetaSize], mv)
			dest[m.Hash] = mv
		}
	}
	return nil
}

var (
	machineId         = []byte(model.LabelMachineId)
	systemUuid        = []byte(model.LabelSystemUuid)
	containerId       = []byte(model.LabelContainerId)
	destination       = []byte(model.LabelDestination)
	destinationIP     = []byte(model.LabelDestinationIP)
	actualDestination = []byte(model.LabelActualDestination)
)

func readLabels(src []byte, mv *model.MetricValues) {
	var key []byte
	isValue := false
	f := 0
	for {
		i := bytes.IndexByte(src[f:], 0)
		if i == -1 {
			break
		}
		i += f
		if isValue {
			if bytes.Equal(key, machineId) {
				mv.MachineID = string(src[f:i])
			} else if bytes.Equal(key, systemUuid) {
				mv.SystemUUID = string(src[f:i])
			} else if bytes.Equal(key, containerId) {
				mv.ContainerId = string(src[f:i])
			} else if bytes.Equal(key, destination) {
				mv.Destination = string(src[f:i])
			} else if bytes.Equal(key, actualDestination) {
				mv.ActualDestination = string(src[f:i])
			} else if bytes.Equal(key, destinationIP) {
				mv.Destination = string(src[f:i])
				mv.DestIp = true
			} else {
				mv.Labels[string(key)] = string(src[f:i])
			}
			isValue = false
		} else {
			key = src[f:i]
			isValue = true
		}
		f = i + 1
	}
}

func asBytes32[T any](slice []T) []byte {
	return unsafe.Slice((*byte)(unsafe.Pointer(&slice[0])), len(slice)*4)
}

func asBytes64[T any](slice []T) []byte {
	return unsafe.Slice((*byte)(unsafe.Pointer(&slice[0])), len(slice)*8)
}

func asFloats32(b []byte) []float32 {
	return unsafe.Slice((*float32)(unsafe.Pointer(&b[0])), len(b)/4)
}

func asUint64(b []byte) []uint64 {
	return unsafe.Slice((*uint64)(unsafe.Pointer(&b[0])), len(b)/8)
}