VictoriaMetrics/app/vmctl/opentsdb/parser.go

175 lines
5.8 KiB
Go
Raw Normal View History

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
}