Merge pull request #26 from raintank/partitionBy

add partitioning of MetricData/MetricDefinitions

Author: Dieterbe

metric.go

@@ -3,9 +3,9 @@ package schema
 import (
 	"bytes"
 	"crypto/md5"
-	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
 	"sort"
 	"strings"
 )
@@ -15,18 +15,13 @@ var ErrInvalidOrgIdzero = errors.New("org-id cannot be 0")
 var ErrInvalidEmptyName = errors.New("name cannot be empty")
 var ErrInvalidMtype = errors.New("invalid mtype")
 var ErrInvalidTagFormat = errors.New("invalid tag format")
+var ErrUnknownPartitionMethod = errors.New("unknown partition method")
 
 type PartitionedMetric interface {
 	Validate() error
 	SetId()
-	// return a []byte key comprised of the metric's OrgId
-	// accepts an input []byte to allow callers to re-use
-	// buffers to reduce memory allocations
-	KeyByOrgId([]byte) []byte
-	// return a []byte key comprised of the metric's Name
-	// accepts an input []byte to allow callers to re-use
-	// buffers to reduce memory allocations
-	KeyBySeries([]byte) []byte
+	// PartitionID returns the partition id that should be used for this metric.
+	PartitionID(method PartitionByMethod, partitions int32) (int32, error)
 }
 
 //go:generate msgp
@@ -63,25 +58,6 @@ func (m *MetricData) Validate() error {
 	return nil
 }
 
-func (m *MetricData) KeyByOrgId(b []byte) []byte {
-	if cap(b)-len(b) < 4 {
-		// not enough unused space in the slice so we need to grow it.
-		newBuf := make([]byte, len(b), len(b)+4)
-		copy(newBuf, b)
-		b = newBuf
-	}
-	// PutUint32 writes directly to the slice rather then appending.
-	// so we need to set the length to 4 more bytes then it currently is.
-	b = b[:len(b)+4]
-	binary.LittleEndian.PutUint32(b[len(b)-4:], uint32(m.OrgId))
-	return b
-}
-
-func (m *MetricData) KeyBySeries(b []byte) []byte {
-	b = append(b, []byte(m.Name)...)
-	return b
-}
-
 // returns a id (hash key) in the format OrgId.md5Sum
 // the md5sum is a hash of the the concatination of the
 // metric + each tag key:value pair (in metrics2.0 sense, so also fields), sorted alphabetically.
@@ -124,7 +100,7 @@ type MetricDefinition struct {
 
 	// this is a special attribute that does not need to be set, it is only used
 	// to keep the state of NameWithTags()
-	nameWithTags string `json:"-"`
+	nameWithTags string
 }
 
 // NameWithTags deduplicates the name and tags strings by storing their content
@@ -137,27 +113,26 @@ func (m *MetricDefinition) NameWithTags() string {
 		return m.nameWithTags
 	}
 
-	sort.Strings(m.Tags)
+	nameWithTagsBuffer := &bytes.Buffer{}
+	_ = writeSortedTagString(nameWithTagsBuffer, m.Name, m.Tags)
+	m.nameWithTags = nameWithTagsBuffer.String()
 
-	nameWithTagsBuffer := bytes.NewBufferString(m.Name)
-	tagPositions := make([]int, 0, len(m.Tags)*2)
+	var i int
+	cursor := len(m.Name)
+	m.Name = m.nameWithTags[:cursor]
 	for _, t := range m.Tags {
-		if len(t) >= 5 && t[:5] == "name=" {
+		if len(t) > 5 && t[:5] == "name=" {
 			continue
 		}
-
-		nameWithTagsBuffer.WriteString(";")
-		tagPositions = append(tagPositions, nameWithTagsBuffer.Len())
-		nameWithTagsBuffer.WriteString(t)
-		tagPositions = append(tagPositions, nameWithTagsBuffer.Len())
+		m.Tags[i] = m.nameWithTags[cursor+1 : cursor+1+len(t)]
+		cursor += len(t) + 1
+		i++
 	}
 
-	m.nameWithTags = nameWithTagsBuffer.String()
-	m.Tags = make([]string, len(tagPositions)/2)
-	for i := 0; i < len(m.Tags); i++ {
-		m.Tags[i] = m.nameWithTags[tagPositions[i*2]:tagPositions[i*2+1]]
+	// if a "name" tag existed, then we have to shorten the slice
+	if i < len(m.Tags) {
+		m.Tags = m.Tags[:i]
 	}
-	m.Name = m.nameWithTags[:len(m.Name)]
 
 	return m.nameWithTags
 }
@@ -178,7 +153,7 @@ func (m *MetricDefinition) SetId() {
 	fmt.Fprintf(buffer, "%d", m.Interval)
 
 	for _, t := range m.Tags {
-		if len(t) >= 5 && t[:5] == "name=" {
+		if len(t) > 5 && t[:5] == "name=" {
 			continue
 		}
 
@@ -211,25 +186,6 @@ func (m *MetricDefinition) Validate() error {
 	return nil
 }
 
-func (m *MetricDefinition) KeyByOrgId(b []byte) []byte {
-	if cap(b)-len(b) < 4 {
-		// not enough unused space in the slice so we need to grow it.
-		newBuf := make([]byte, len(b), len(b)+4)
-		copy(newBuf, b)
-		b = newBuf
-	}
-	// PutUint32 writes directly to the slice rather then appending.
-	// so we need to set the length to 4 more bytes then it currently is.
-	b = b[:len(b)+4]
-	binary.LittleEndian.PutUint32(b[len(b)-4:], uint32(m.OrgId))
-	return b
-}
-
-func (m *MetricDefinition) KeyBySeries(b []byte) []byte {
-	b = append(b, []byte(m.Name)...)
-	return b
-}
-
 // MetricDefinitionFromMetricData yields a MetricDefinition that has no references
 // to the original MetricData
 func MetricDefinitionFromMetricData(d *MetricData) *MetricDefinition {
@@ -327,3 +283,30 @@ func ValidateTagValue(value string) bool {
 
 	return !strings.ContainsRune(value, ';')
 }
+
+func writeSortedTagString(w io.Writer, name string, tags []string) error {
+	sort.Strings(tags)
+
+	_, err := io.WriteString(w, name)
+	if err != nil {
+		return err
+	}
+
+	for _, t := range tags {
+		if len(t) > 5 && t[:5] == "name=" {
+			continue
+		}
+
+		_, err = io.WriteString(w, ";")
+		if err != nil {
+			return err
+		}
+
+		_, err = io.WriteString(w, t)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

partition.go

@@ -0,0 +1,116 @@
+package schema
+
+import (
+	"encoding/binary"
+	"hash/fnv"
+
+	"github.com/cespare/xxhash"
+	jump "github.com/dgryski/go-jump"
+)
+
+type PartitionByMethod uint8
+
+const (
+	// partition by organization id only
+	PartitionByOrg PartitionByMethod = iota
+
+	// partition by the metric name only
+	PartitionBySeries
+
+	// partition by metric name and tags, with the best distribution
+	// recommended for new deployments.
+	PartitionBySeriesWithTags
+
+	// partition by metric name and tags, with a sub-optimal distribution when using tags.
+	// compatible with PartitionBySeries if a metric has no tags,
+	// making it possible to adopt tags for existing PartitionBySeries deployments without a migration.
+	PartitionBySeriesWithTagsFnv
+)
+
+func (m *MetricData) PartitionID(method PartitionByMethod, partitions int32) (int32, error) {
+	var partition int32
+
+	switch method {
+	case PartitionByOrg:
+		h := fnv.New32a()
+		err := binary.Write(h, binary.LittleEndian, uint32(m.OrgId))
+		if err != nil {
+			return 0, err
+		}
+		partition = int32(h.Sum32()) % partitions
+		if partition < 0 {
+			partition = -partition
+		}
+	case PartitionBySeries:
+		h := fnv.New32a()
+		h.Write([]byte(m.Name))
+		partition = int32(h.Sum32()) % partitions
+		if partition < 0 {
+			partition = -partition
+		}
+	case PartitionBySeriesWithTags:
+		h := xxhash.New()
+		if err := writeSortedTagString(h, m.Name, m.Tags); err != nil {
+			return 0, err
+		}
+		partition = jump.Hash(h.Sum64(), int(partitions))
+	case PartitionBySeriesWithTagsFnv:
+		h := fnv.New32a()
+		if err := writeSortedTagString(h, m.Name, m.Tags); err != nil {
+			return 0, err
+		}
+		partition = int32(h.Sum32()) % partitions
+		if partition < 0 {
+			partition = -partition
+		}
+	default:
+		return 0, ErrUnknownPartitionMethod
+	}
+
+	return partition, nil
+}
+
+func (m *MetricDefinition) PartitionID(method PartitionByMethod, partitions int32) (int32, error) {
+	var partition int32
+
+	switch method {
+	case PartitionByOrg:
+		h := fnv.New32a()
+		err := binary.Write(h, binary.LittleEndian, uint32(m.OrgId))
+		if err != nil {
+			return 0, err
+		}
+		partition = int32(h.Sum32()) % partitions
+		if partition < 0 {
+			partition = -partition
+		}
+	case PartitionBySeries:
+		h := fnv.New32a()
+		h.Write([]byte(m.Name))
+		partition = int32(h.Sum32()) % partitions
+		if partition < 0 {
+			partition = -partition
+		}
+	case PartitionBySeriesWithTags:
+		h := xxhash.New()
+		h.WriteString(m.NameWithTags())
+		partition = jump.Hash(h.Sum64(), int(partitions))
+	case PartitionBySeriesWithTagsFnv:
+		h := fnv.New32a()
+		if len(m.nameWithTags) > 0 {
+			h.Write([]byte(m.nameWithTags))
+		} else {
+			if err := writeSortedTagString(h, m.Name, m.Tags); err != nil {
+				return 0, err
+			}
+		}
+		partition = int32(h.Sum32()) % partitions
+		if partition < 0 {
+			partition = -partition
+		}
+	default:
+		return 0, ErrUnknownPartitionMethod
+	}
+
+	return partition, nil
+}