mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2024-12-01 14:47:38 +00:00
174 lines
5.8 KiB
Go
174 lines
5.8 KiB
Go
|
package opentsdb
|
||
|
|
||
|
import (
|
||
|
"fmt"
|
||
|
"regexp"
|
||
|
"strconv"
|
||
|
"strings"
|
||
|
"time"
|
||
|
)
|
||
|
|
||
|
var (
|
||
|
allowedNames = regexp.MustCompile("^[a-zA-Z][a-zA-Z0-9_:]*$")
|
||
|
allowedFirstChar = regexp.MustCompile("^[a-zA-Z]")
|
||
|
replaceChars = regexp.MustCompile("[^a-zA-Z0-9_:]")
|
||
|
allowedTagKeys = regexp.MustCompile("^[a-zA-Z][a-zA-Z0-9_]*$")
|
||
|
)
|
||
|
|
||
|
func convertDuration(duration string) (time.Duration, error) {
|
||
|
/*
|
||
|
Golang's time library doesn't support many different
|
||
|
string formats (year, month, week, day) because they
|
||
|
aren't consistent ranges. But Java's library _does_.
|
||
|
Consequently, we'll need to handle all the custom
|
||
|
time ranges, and, to make the internal API call consistent,
|
||
|
we'll need to allow for durations that Go supports, too.
|
||
|
|
||
|
The nice thing is all the "broken" time ranges are > 1 hour,
|
||
|
so we can just make assumptions to convert them to a range in hours.
|
||
|
They aren't *good* assumptions, but they're reasonable
|
||
|
for this function.
|
||
|
*/
|
||
|
var actualDuration time.Duration
|
||
|
var err error
|
||
|
var timeValue int
|
||
|
if strings.HasSuffix(duration, "y") {
|
||
|
timeValue, err = strconv.Atoi(strings.Trim(duration, "y"))
|
||
|
if err != nil {
|
||
|
return 0, fmt.Errorf("invalid time range: %q", duration)
|
||
|
}
|
||
|
timeValue = timeValue * 365 * 24
|
||
|
actualDuration, err = time.ParseDuration(fmt.Sprintf("%vh", timeValue))
|
||
|
if err != nil {
|
||
|
return 0, fmt.Errorf("invalid time range: %q", duration)
|
||
|
}
|
||
|
} else if strings.HasSuffix(duration, "w") {
|
||
|
timeValue, err = strconv.Atoi(strings.Trim(duration, "w"))
|
||
|
if err != nil {
|
||
|
return 0, fmt.Errorf("invalid time range: %q", duration)
|
||
|
}
|
||
|
timeValue = timeValue * 7 * 24
|
||
|
actualDuration, err = time.ParseDuration(fmt.Sprintf("%vh", timeValue))
|
||
|
if err != nil {
|
||
|
return 0, fmt.Errorf("invalid time range: %q", duration)
|
||
|
}
|
||
|
} else if strings.HasSuffix(duration, "d") {
|
||
|
timeValue, err = strconv.Atoi(strings.Trim(duration, "d"))
|
||
|
if err != nil {
|
||
|
return 0, fmt.Errorf("invalid time range: %q", duration)
|
||
|
}
|
||
|
timeValue = timeValue * 24
|
||
|
actualDuration, err = time.ParseDuration(fmt.Sprintf("%vh", timeValue))
|
||
|
if err != nil {
|
||
|
return 0, fmt.Errorf("invalid time range: %q", duration)
|
||
|
}
|
||
|
} else if strings.HasSuffix(duration, "h") || strings.HasSuffix(duration, "m") || strings.HasSuffix(duration, "s") || strings.HasSuffix(duration, "ms") {
|
||
|
actualDuration, err = time.ParseDuration(duration)
|
||
|
if err != nil {
|
||
|
return 0, fmt.Errorf("invalid time range: %q", duration)
|
||
|
}
|
||
|
} else {
|
||
|
return 0, fmt.Errorf("invalid time duration string: %q", duration)
|
||
|
}
|
||
|
return actualDuration, nil
|
||
|
}
|
||
|
|
||
|
// Convert an incoming retention "string" into the component parts
|
||
|
func convertRetention(retention string, offset int64, msecTime bool) (Retention, error) {
|
||
|
/*
|
||
|
A retention string coming in looks like
|
||
|
sum-1m-avg:1h:30d
|
||
|
So we:
|
||
|
1. split on the :
|
||
|
2. split on the - in slice 0
|
||
|
3. create the time ranges we actually need
|
||
|
*/
|
||
|
chunks := strings.Split(retention, ":")
|
||
|
if len(chunks) != 3 {
|
||
|
return Retention{}, fmt.Errorf("invalid retention string: %q", retention)
|
||
|
}
|
||
|
rowLengthDuration, err := convertDuration(chunks[1])
|
||
|
if err != nil {
|
||
|
return Retention{}, fmt.Errorf("invalid row length (first order) duration string: %q: %s", chunks[1], err)
|
||
|
}
|
||
|
// set length of each row in milliseconds, unless we aren't using millisecond time in OpenTSDB...then use seconds
|
||
|
rowLength := rowLengthDuration.Milliseconds()
|
||
|
if !msecTime {
|
||
|
rowLength = rowLength / 1000
|
||
|
}
|
||
|
ttlDuration, err := convertDuration(chunks[2])
|
||
|
if err != nil {
|
||
|
return Retention{}, fmt.Errorf("invalid ttl (second order) duration string: %q: %s", chunks[2], err)
|
||
|
}
|
||
|
// set ttl in milliseconds, unless we aren't using millisecond time in OpenTSDB...then use seconds
|
||
|
ttl := ttlDuration.Milliseconds()
|
||
|
if !msecTime {
|
||
|
ttl = ttl / 1000
|
||
|
}
|
||
|
// bump by the offset so we don't look at empty ranges any time offset > ttl
|
||
|
ttl += offset
|
||
|
var timeChunks []TimeRange
|
||
|
var i int64
|
||
|
for i = offset; i <= ttl; i = i + rowLength {
|
||
|
timeChunks = append(timeChunks, TimeRange{Start: i + rowLength, End: i})
|
||
|
}
|
||
|
// first/second order aggregations for queries defined in chunk 0...
|
||
|
aggregates := strings.Split(chunks[0], "-")
|
||
|
if len(aggregates) != 3 {
|
||
|
return Retention{}, fmt.Errorf("invalid aggregation string: %q", chunks[0])
|
||
|
}
|
||
|
|
||
|
ret := Retention{FirstOrder: aggregates[0],
|
||
|
SecondOrder: aggregates[2],
|
||
|
AggTime: aggregates[1],
|
||
|
QueryRanges: timeChunks}
|
||
|
return ret, nil
|
||
|
}
|
||
|
|
||
|
// This ensures any incoming data from OpenTSDB matches the Prometheus data model
|
||
|
// https://prometheus.io/docs/concepts/data_model
|
||
|
func modifyData(msg Metric, normalize bool) (Metric, error) {
|
||
|
finalMsg := Metric{
|
||
|
Metric: "", Tags: make(map[string]string),
|
||
|
Timestamps: msg.Timestamps, Values: msg.Values,
|
||
|
}
|
||
|
// if the metric name has invalid characters, the data model says to drop it
|
||
|
if !allowedFirstChar.MatchString(msg.Metric) {
|
||
|
return Metric{}, fmt.Errorf("%s has a bad first character", msg.Metric)
|
||
|
}
|
||
|
name := msg.Metric
|
||
|
// if normalization requested, lowercase the name
|
||
|
if normalize {
|
||
|
name = strings.ToLower(name)
|
||
|
}
|
||
|
/*
|
||
|
replace bad characters in metric name with _ per the data model
|
||
|
only replace if needed to reduce string processing time
|
||
|
*/
|
||
|
if !allowedNames.MatchString(name) {
|
||
|
finalMsg.Metric = replaceChars.ReplaceAllString(name, "_")
|
||
|
} else {
|
||
|
finalMsg.Metric = name
|
||
|
}
|
||
|
// replace bad characters in tag keys with _ per the data model
|
||
|
for key, value := range msg.Tags {
|
||
|
// if normalization requested, lowercase the key and value
|
||
|
if normalize {
|
||
|
key = strings.ToLower(key)
|
||
|
value = strings.ToLower(value)
|
||
|
}
|
||
|
/*
|
||
|
replace all explicitly bad characters with _
|
||
|
only replace if needed to reduce string processing time
|
||
|
*/
|
||
|
if !allowedTagKeys.MatchString(key) {
|
||
|
key = replaceChars.ReplaceAllString(key, "_")
|
||
|
}
|
||
|
// tags that start with __ are considered custom stats for internal prometheus stuff, we should drop them
|
||
|
if !strings.HasPrefix(key, "__") {
|
||
|
finalMsg.Tags[key] = value
|
||
|
}
|
||
|
}
|
||
|
return finalMsg, nil
|
||
|
}
|