Golang bosun-org-opentsdb.ParseTags类(方法)实例源码

// parseTcollectorValue parses a tcollector-style line into a data point.
func parseTcollectorValue(line string) (*opentsdb.DataPoint, error) {
	sp := strings.Fields(line)
	if len(sp) < 3 {
		return nil, fmt.Errorf("bad line: %s", line)
	}
	ts, err := strconv.ParseInt(sp[1], 10, 64)
	if err != nil {
		return nil, fmt.Errorf("bad timestamp: %s", sp[1])
	}
	val, err := strconv.ParseFloat(sp[2], 64)
	if err != nil {
		return nil, fmt.Errorf("bad value: %s", sp[2])
	}
	if !opentsdb.ValidTag(sp[0]) {
		return nil, fmt.Errorf("bad metric: %s", sp[0])
	}
	dp := opentsdb.DataPoint{
		Metric:    sp[0],
		Timestamp: ts,
		Value:     val,
	}
	tags := opentsdb.TagSet{}
	for _, tag := range sp[3:] {
		ts, err := opentsdb.ParseTags(tag)
		if err != nil {
			return nil, fmt.Errorf("bad tag, metric %s: %v: %v", sp[0], tag, err)
		}
		tags.Merge(ts)
	}
	setExternalTags(tags)
	dp.Tags = tags
	return &dp, nil
}

func (c *Context) LookupAll(table, key string, group interface{}) string {
	var t opentsdb.TagSet
	switch v := group.(type) {
	case string:
		var err error
		t, err = opentsdb.ParseTags(v)
		if err != nil {
			c.addError(err)
			return err.Error()
		}
	case opentsdb.TagSet:
		t = v
	}
	l := c.schedule.RuleConf.GetLookup(table)
	if l == nil {
		err := fmt.Errorf("unknown lookup table %v", table)
		c.addError(err)
		return err.Error()
	}
	if v, ok := l.ToExpr().Get(key, t); ok {
		return v
	}
	err := fmt.Errorf("no entry for key %v in table %v for tagset %v", key, table, c.AlertKey.Group())
	c.addError(err)
	return err.Error()
}

func (c *Context) GetMeta(metric, name string, v interface{}) interface{} {
	var t opentsdb.TagSet
	switch v := v.(type) {
	case string:
		var err error
		t, err = opentsdb.ParseTags(v)
		if err != nil {
			c.addError(err)
			return nil
		}
	case opentsdb.TagSet:
		t = v
	}
	meta, err := c.schedule.GetMetadata(metric, t)
	if err != nil {
		c.addError(err)
		return nil
	}
	if name == "" {
		return meta
	}
	for _, m := range meta {
		if m.Name == name {
			return m.Value
		}
	}
	return "metadta not found"
}

func (c *Context) GetMeta(metric, name string, v interface{}) (interface{}, error) {
	var t opentsdb.TagSet
	switch v := v.(type) {
	case string:
		var err error
		t, err = opentsdb.ParseTags(v)
		if err != nil {
			return t, err
		}
	case opentsdb.TagSet:
		t = v
	}
	meta, err := c.schedule.GetMetadata(metric, t)
	if err != nil {
		return nil, err
	}
	if name == "" {
		return meta, nil
	}
	for _, m := range meta {
		if m.Name == name {
			return m.Value, nil
		}
	}
	return nil, nil
}

func SeriesFunc(e *State, T miniprofiler.Timer, tags string, pairs ...float64) (*Results, error) {
	if len(pairs)%2 != 0 {
		return nil, fmt.Errorf("uneven number of time stamps and values")
	}
	group := opentsdb.TagSet{}
	if tags != "" {
		var err error
		group, err = opentsdb.ParseTags(tags)
		if err != nil {
			return nil, fmt.Errorf("unable to parse tags: %v", err)
		}
	}

	series := make(Series)
	for i := 0; i < len(pairs); i += 2 {
		series[time.Unix(int64(pairs[i]), 0)] = pairs[i+1]
	}
	return &Results{
		Results: []*Result{
			{
				Value: series,
				Group: group,
			},
		},
	}, nil
}

// TagSet returns m's tags.
func (m Metakey) TagSet() opentsdb.TagSet {
	tags, err := opentsdb.ParseTags(m.Tags)
	if err != nil {
		return nil
	}
	return tags
}

func (s *Schedule) AddSilence(start, end time.Time, alert, tagList string, forget, confirm bool, edit, user, message string) (map[models.AlertKey]bool, error) {
	if start.IsZero() || end.IsZero() {
		return nil, fmt.Errorf("both start and end must be specified")
	}
	if start.After(end) {
		return nil, fmt.Errorf("start time must be before end time")
	}
	if time.Since(end) > 0 {
		return nil, fmt.Errorf("end time must be in the future")
	}
	if alert == "" && tagList == "" {
		return nil, fmt.Errorf("must specify either alert or tags")
	}
	si := &models.Silence{
		Start:   start,
		End:     end,
		Alert:   alert,
		Tags:    make(opentsdb.TagSet),
		Forget:  forget,
		User:    user,
		Message: message,
	}
	if tagList != "" {
		tags, err := opentsdb.ParseTags(tagList)
		if err != nil && tags == nil {
			return nil, err
		}
		si.Tags = tags
		si.TagString = tags.Tags()
	}
	if confirm {
		if edit != "" {
			if err := s.DataAccess.Silence().DeleteSilence(edit); err != nil {
				return nil, err
			}
		}
		if err := s.DataAccess.Silence().DeleteSilence(si.ID()); err != nil {
			return nil, err
		}
		if err := s.DataAccess.Silence().AddSilence(si); err != nil {
			return nil, err
		}
		return nil, nil
	}
	aks := make(map[models.AlertKey]bool)
	open, err := s.DataAccess.State().GetAllOpenIncidents()
	if err != nil {
		return nil, err
	}
	for _, inc := range open {
		if si.Matches(inc.Alert, inc.AlertKey.Group()) {
			aks[inc.AlertKey] = true
		}
	}
	return aks, nil
}

// Last returns the most recent datapoint for a metric+tagset. The metric+tagset
// string should be formated like os.cpu{host=foo}. The tag porition expects the
// that the keys will be in alphabetical order.
func Last(t miniprofiler.Timer, w http.ResponseWriter, r *http.Request) (interface{}, error) {
	var counter bool
	if r.FormValue("counter") != "" {
		counter = true
	}
	tags, err := opentsdb.ParseTags(r.FormValue("tagset"))
	if err != nil {
		return nil, err
	}
	return schedule.Search.GetLast(r.FormValue("metric"), tags, counter)
}

func c_redis_counters(server string, db int) (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	conn, err := redis.Dial("tcp", server, redis.DialDatabase(db))
	if err != nil {
		return md, err
	}
	defer conn.Close()
	if _, err := conn.Do("CLIENT", "SETNAME", "scollector"); err != nil {
		return md, err
	}
	cursor := 0
	for {
		vals, err := redis.Values(conn.Do("HSCAN", collect.RedisCountersKey, cursor))
		if err != nil {
			return md, err
		}
		if len(vals) != 2 {
			return md, fmt.Errorf("Unexpected number of values")
		}
		cursor, err = redis.Int(vals[0], nil)
		if err != nil {
			return md, err
		}
		pairs, err := redis.StringMap(vals[1], nil)
		if err != nil {
			return md, err
		}
		for mts, val := range pairs {
			parts := strings.Split(mts, ":")
			if len(parts) != 2 {
				slog.Errorf("Invalid metric tag set counter: %s", mts)
				continue
			}
			metric := parts[0]
			tags, err := opentsdb.ParseTags(parts[1])
			if err != nil {
				slog.Errorf("Invalid tags: %s", parts[1])
				continue
			}
			v, err := strconv.Atoi(val)
			if err != nil {
				slog.Errorf("Invalid counter value: %s", val)
				continue
			}
			Add(&md, metric, v, tags, metadata.Counter, metadata.Count, "")
		}
		if cursor == 0 {
			break
		}
	}
	return md, nil
}

func FilteredTagsetsByMetric(t miniprofiler.Timer, w http.ResponseWriter, r *http.Request) (interface{}, error) {
	vars := mux.Vars(r)
	metric := vars["metric"]
	tagset := opentsdb.TagSet{}
	var err error
	ts := r.FormValue("tags")
	if ts != "" {
		if tagset, err = opentsdb.ParseTags(ts); err != nil {
			return nil, err
		}
	}
	return schedule.Search.FilteredTagSets(metric, tagset)
}

func (d *dataAccess) GetTagMetadata(tags opentsdb.TagSet, name string) ([]*TagMetadata, error) {
	defer collect.StartTimer("redis", opentsdb.TagSet{"op": "GetTagMeta"})()
	conn := d.GetConnection()
	defer conn.Close()
	args := []interface{}{}
	for tagK, tagV := range tags {
		args = append(args, tagMetaIdxKey(tagK, tagV))
	}
	keys, err := redis.Strings(conn.Do("SINTER", args...))
	if err != nil {
		return nil, err
	}
	args = []interface{}{}
	for _, key := range keys {
		if name == "" || strings.HasSuffix(key, ":"+name) {
			args = append(args, key)
		}
	}
	results, err := redis.Strings(conn.Do("MGET", args...))
	data := []*TagMetadata{}
	for i := range args {
		// break up key to get tags and name
		key := args[i].(string)[len("tmeta:"):]
		sepIdx := strings.LastIndex(key, ":")
		tags := key[:sepIdx]
		name := key[sepIdx+1:]
		tagSet, err := opentsdb.ParseTags(tags)
		if err != nil {
			return nil, err
		}
		// break up response to get time and value
		parts := strings.SplitN(results[i], ":", 2)
		if len(parts) != 2 {
			return nil, fmt.Errorf("Expect metadata value to be `time:value`")
		}
		val := parts[1]
		time, err := strconv.ParseInt(parts[0], 10, 64)
		if err != nil {
			return nil, err
		}
		obj := &TagMetadata{
			Tags:        tagSet,
			Name:        name,
			Value:       val,
			LastTouched: time,
		}
		data = append(data, obj)
	}
	return data, nil
}

func (s *Search) FilteredTagSets(metric string, tags opentsdb.TagSet) ([]opentsdb.TagSet, error) {
	sets, err := s.DataAccess.Search().GetMetricTagSets(metric, tags)
	if err != nil {
		return nil, err
	}
	r := []opentsdb.TagSet{}
	for k := range sets {
		ts, err := opentsdb.ParseTags(k)
		if err != nil {
			return nil, err
		}
		r = append(r, ts)
	}
	return r, nil
}

// Group returns the tagset of this alert key. Will panic if a is not a valid
// AlertKey. OpenTSDB tag validation errors are ignored.
func (a AlertKey) Group() opentsdb.TagSet {
	sp := strings.SplitN(string(a), "{", 2)
	if len(sp) < 2 {
		panic(fmt.Errorf("invalid alert key %s", a))
	}
	s := sp[1]
	s = s[:len(s)-1]
	if s == "" {
		return nil
	}
	g, err := opentsdb.ParseTags(s)
	if g == nil && err != nil {
		panic(err)
	}
	return g
}

func (s *Squelches) Add(v string) error {
	tags, err := opentsdb.ParseTags(v)
	if tags == nil && err != nil {
		return err
	}
	sq := make(Squelch)
	for k, v := range tags {
		re, err := regexp.Compile(v)
		if err != nil {
			return err
		}
		sq[k] = re
	}
	s.s = append(s.s, sq)
	return nil
}

func (s *Schedule) AddSilence(start, end time.Time, alert, tagList string, forget, confirm bool, edit, user, message string) (map[expr.AlertKey]bool, error) {
	if start.IsZero() || end.IsZero() {
		return nil, fmt.Errorf("both start and end must be specified")
	}
	if start.After(end) {
		return nil, fmt.Errorf("start time must be before end time")
	}
	if time.Since(end) > 0 {
		return nil, fmt.Errorf("end time must be in the future")
	}
	if alert == "" && tagList == "" {
		return nil, fmt.Errorf("must specify either alert or tags")
	}
	si := &Silence{
		Start:   start,
		End:     end,
		Alert:   alert,
		Tags:    make(opentsdb.TagSet),
		Forget:  forget,
		User:    user,
		Message: message,
	}
	if tagList != "" {
		tags, err := opentsdb.ParseTags(tagList)
		if err != nil && tags == nil {
			return nil, err
		}
		si.Tags = tags
	}
	s.Lock("AddSilence")
	defer s.Unlock()
	if confirm {
		delete(s.Silence, edit)
		s.Silence[si.ID()] = si
		s.Save()
		return nil, nil
	}
	aks := make(map[expr.AlertKey]bool)
	for ak := range s.status {
		if si.Matches(ak.Name(), ak.Group()) {
			aks[ak] = s.status[ak].IsActive()
		}
	}
	return aks, nil
}

func AddTags(e *State, T miniprofiler.Timer, series *Results, s string) (*Results, error) {
	if s == "" {
		return series, nil
	}
	tagSetToAdd, err := opentsdb.ParseTags(s)
	if err != nil {
		return nil, err
	}
	for tagKey, tagValue := range tagSetToAdd {
		for _, res := range series.Results {
			if _, ok := res.Group[tagKey]; ok {
				return nil, fmt.Errorf("%s key already in group", tagKey)
			}
			res.Group[tagKey] = tagValue
		}
	}
	return series, nil
}

func (d *dataAccess) Search_GetMetricTagSets(metric string, tags opentsdb.TagSet) (map[string]int64, error) {
	defer collect.StartTimer("redis", opentsdb.TagSet{"op": "GetMetricTagSets"})()
	conn := d.GetConnection()
	defer conn.Close()

	mtss, err := stringInt64Map(conn.Do("HGETALL", searchMetricTagSetKey(metric)))
	if err != nil {
		return nil, err
	}
	for mts := range mtss {
		ts, err := opentsdb.ParseTags(mts)
		if err != nil {
			return nil, err
		}
		if !ts.Subset(tags) {
			delete(mtss, mts)
		}
	}
	return mtss, nil
}

func (c *Context) LookupAll(table, key string, group interface{}) (string, error) {
	var t opentsdb.TagSet
	switch v := group.(type) {
	case string:
		var err error
		t, err = opentsdb.ParseTags(v)
		if err != nil {
			return "", err
		}
	case opentsdb.TagSet:
		t = v
	}
	l, ok := c.schedule.Conf.Lookups[table]
	if !ok {
		return "", fmt.Errorf("unknown lookup table %v", table)
	}
	if v, ok := l.ToExpr().Get(key, t); ok {
		return v, nil
	}
	return "", fmt.Errorf("no entry for key %v in table %v for tagset %v", key, table, c.Group)
}

func FilteredTagsetsByMetric(t miniprofiler.Timer, w http.ResponseWriter, r *http.Request) (interface{}, error) {
	vars := mux.Vars(r)
	metric := vars["metric"]
	tagset := opentsdb.TagSet{}
	var err error
	ts := r.FormValue("tags")
	if ts != "" {
		if tagset, err = opentsdb.ParseTags(ts); err != nil {
			return nil, err
		}
	}

	since := int64(0)
	sinceStr := r.FormValue("since")
	if sinceStr != "" {
		since, err = strconv.ParseInt(sinceStr, 10, 64) //since will be set to 0 again in case of errors
		if err != nil {
			return nil, err
		}
	}
	return schedule.Search.FilteredTagSets(metric, tagset, since)
}

func (d *dataAccess) GetMetricTagSets(metric string, tags opentsdb.TagSet) (map[string]int64, error) {
	defer collect.StartTimer("redis", opentsdb.TagSet{"op": "GetMetricTagSets"})()
	conn := d.GetConnection()
	defer conn.Close()

	var cursor = "0"
	result := map[string]int64{}

	for {
		vals, err := redis.Values(conn.Do(d.HSCAN(), searchMetricTagSetKey(metric), cursor))
		if err != nil {
			return nil, slog.Wrap(err)
		}
		cursor, err = redis.String(vals[0], nil)
		if err != nil {
			return nil, slog.Wrap(err)
		}
		mtss, err := stringInt64Map(vals[1], nil)
		if err != nil {
			return nil, slog.Wrap(err)
		}
		for mts, t := range mtss {
			ts, err := opentsdb.ParseTags(mts)
			if err != nil {
				return nil, slog.Wrap(err)
			}
			if ts.Subset(tags) {
				result[mts] = t
			}
		}

		if cursor == "" || cursor == "0" {
			break
		}
	}
	return result, nil
}