Golang bosun-org-metadata.AddMeta类(方法)实例源码

func extraHopGetCertificateByCount(c *gohop.Client, md *opentsdb.MultiDataPoint) error {
	//These are the metrics we are populating in this part of the collector
	metricNameCount := "extrahop.certificates"

	//Metadata for the above metrics
	metadata.AddMeta(metricNameCount, nil, "rate", metadata.Gauge, false)
	metadata.AddMeta(metricNameCount, nil, "unit", metadata.Count, false)
	metadata.AddMeta(metricNameCount, nil, "desc", "The number of times a given certificate was seen", false)

	ms := []gohop.MetricSpec{ //Build a metric spec to tell ExtraHop what we want to pull out.
		{Name: "cert_subject", KeyPair: gohop.KeyPair{Key1Regex: "", Key2Regex: "", OpenTSDBKey1: "", Key2OpenTSDBKey2: ""}, OpenTSDBMetric: metricNameCount},
	}
	mrk, err := c.KeyedMetricQuery(gohop.Cycle30Sec, "ssl_server_detail", "activity_group", -60000, 0, ms, []int64{int64(extraHopCertificateActivityGroup)})
	if err != nil {
		return err
	}

	//At this time we have a keyed metric response from ExtraHop. We need to find all the stats, then the values of the stats, and then
	//filter out to only the records we want.

	//This is our function that is going to be executed on each data point in the extrahop dataset
	appendCountPoints := func(c *gohop.Client, md *opentsdb.MultiDataPoint, a *gohop.MetricStatKeyed, b *[]gohop.MetricStatKeyedValue, d *gohop.MetricStatKeyedValue) {
		thisPoint := getSSLDataPointFromSet(metricNameCount, c.APIUrl.Host, a.Time, d)
		if thisPoint != nil {
			*md = append(*md, thisPoint)
		}
	}

	processGohopStat(&mrk, c, md, appendCountPoints) //This will loop through our datapoint structure and execute appendCountPoints on each final data piece

	return nil
}

// AddTS is the same as Add but lets you specify the timestamp
func AddTS(md *opentsdb.MultiDataPoint, name string, ts int64, value interface{}, t opentsdb.TagSet, rate metadata.RateType, unit metadata.Unit, desc string) {
	if b, ok := value.(bool); ok {
		if b {
			value = 1
		} else {
			value = 0
		}
	}
	tags := t.Copy()
	if host, present := tags["host"]; !present {
		tags["host"] = util.Hostname
	} else if host == "" {
		delete(tags, "host")
	}
	if rate != metadata.Unknown {
		metadata.AddMeta(name, nil, "rate", rate, false)
	}
	if unit != metadata.None {
		metadata.AddMeta(name, nil, "unit", unit, false)
	}
	if desc != "" {
		metadata.AddMeta(name, tags, "desc", desc, false)
	}

	tags = AddTags.Copy().Merge(tags)
	d := opentsdb.DataPoint{
		Metric:    name,
		Timestamp: ts,
		Value:     value,
		Tags:      tags,
	}
	*md = append(*md, &d)
}

func c_meta_darwin_version() (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	util.ReadCommand(func(line string) error {
		metadata.AddMeta("", nil, "uname", line, true)
		return nil
	}, "uname", "-a")
	var name, vers, build string
	util.ReadCommand(func(line string) error {
		sp := strings.SplitN(line, ":", 2)
		if len(sp) != 2 {
			return nil
		}
		v := strings.TrimSpace(sp[1])
		switch sp[0] {
		case "ProductName":
			name = v
		case "ProductVersion":
			vers = v
		case "BuildVersion":
			build = v
		}
		return nil
	}, "sw_vers")
	if name != "" && vers != "" && build != "" {
		metadata.AddMeta("", nil, "version", fmt.Sprintf("%s.%s", vers, build), true)
		metadata.AddMeta("", nil, "versionCaption", fmt.Sprintf("%s %s", name, vers), true)
	}
	return md, nil
}

// ExtraHop collection registration
func ExtraHop(host, apikey, filterby string, filterpercent int, customMetrics []string, certMatch string, certActivityGroup int) error {
	if host == "" || apikey == "" {
		return fmt.Errorf("Empty host or API key for ExtraHop.")
	}

	extraHopAdditionalMetrics = customMetrics
	extraHopFilterProtoBy = filterby
	switch filterby { //Set up options
	case "toppercent":
		extraHopL7Description = fmt.Sprintf("Only the top %d percent of traffic has its protocols logged, the remainder is tagged as as proto=otherprotos", extraHopTopProtoPerc)
		extraHopOtherProtoName = "otherprotos"
		if filterpercent > 0 && filterpercent < 100 {
			extraHopTopProtoPerc = filterpercent
		} else {
			return fmt.Errorf("Invalid ExtraHop FilterPercent value (%d). Number should be between 1 and 99.", filterpercent)
		}
	case "namedprotocols":
		extraHopL7Description = "Only named protocols are logged. Any unnamed protocol (A protocol name starting with tcp, udp or ssl) is tagged as proto=unnamed"
		extraHopOtherProtoName = "unnamed"
	//There is also case "none", but in that case the options we need to keep as default, so there's actually nothing to do here.
	default:
		return fmt.Errorf("Invalid ExtraHop FilterBy option (%s). Valid options are namedprotocols, toppercent or none.", filterby)

	}
	//Add the metadata for the L7 types, as now we have enough information to know what they're going to be
	for l7type, l7s := range l7types {
		xhMetricName := fmt.Sprintf("extrahop.l7.%s", l7type)
		metadata.AddMeta(xhMetricName, nil, "rate", l7s.Rate, false)
		metadata.AddMeta(xhMetricName, nil, "unit", l7s.Unit, false)
		metadata.AddMeta(xhMetricName, nil, "desc", fmt.Sprintf("%s %s", l7s.Description, extraHopL7Description), false)
	}
	u, err := url.Parse(host)
	if err != nil {
		return err
	}

	if certMatch != "" {
		compiledRegexp, err := regexp.Compile(certMatch)
		if err != nil {
			return err
		}
		extraHopCertificateMatch = compiledRegexp
		extraHopCertificateActivityGroup = certActivityGroup
	}
	collectors = append(collectors, &IntervalCollector{
		F: func() (opentsdb.MultiDataPoint, error) {
			return c_extrahop(host, apikey)
		},
		name:     fmt.Sprintf("extrahop-%s", u.Host),
		Interval: time.Second * time.Duration(extraHopIntervalSeconds),
	})
	return nil

}

// AddTS is the same as Add but lets you specify the timestamp
func AddTS(md *opentsdb.MultiDataPoint, name string, ts int64, value interface{}, t opentsdb.TagSet, rate metadata.RateType, unit metadata.Unit, desc string) {
	// Check if we really want that metric
	if skipMetric(name) {
		return
	}

	tags := t.Copy()
	if host, present := tags["host"]; !present {
		tags["host"] = util.Hostname
	} else if host == "" {
		delete(tags, "host")
	}
	// if tags are not cleanable, log a message and skip it
	if err := tags.Clean(); err != nil {
		line := ""
		//attempt to log where Add was called from
		if _, filename, l, ok := runtime.Caller(1); ok {
			if filepath.Base(filename) == "collectors.go" {
				_, filename, l, ok = runtime.Caller(2)
			}
			if ok {
				line = fmt.Sprintf("%s:%d", filepath.Base(filename), l)
			}
		}
		slog.Errorf("Invalid tagset discovered: %s. Skipping datapoint. Added from: %s", tags.String(), line)
		return
	}
	if rate != metadata.Unknown {
		metadata.AddMeta(name, nil, "rate", rate, false)
	}
	if unit != metadata.None {
		metadata.AddMeta(name, nil, "unit", unit, false)
	}
	if desc != "" {
		metadata.AddMeta(name, tags, "desc", desc, false)
	}
	tags = AddTags.Copy().Merge(tags)
	if b, ok := value.(bool); ok {
		if b {
			value = 1
		} else {
			value = 0
		}
	}
	d := opentsdb.DataPoint{
		Metric:    name,
		Timestamp: ts,
		Value:     value,
		Tags:      tags,
	}
	*md = append(*md, &d)
}

func c_omreport_ps() (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	readOmreport(func(fields []string) {
		if len(fields) < 3 || fields[0] == "Index" {
			return
		}
		id := strings.Replace(fields[0], ":", "_", -1)
		ts := opentsdb.TagSet{"id": id}
		Add(&md, "hw.ps", severity(fields[1]), ts, metadata.Gauge, metadata.Ok, descDellHWPS)
		pm := &metadata.HWPowerSupply{}
		if len(fields) < 6 {
			return
		}
		if fields[4] != "" {
			pm.RatedInputWattage = fields[4]
		}
		if fields[5] != "" {
			pm.RatedOutputWattage = fields[5]
		}
		if j, err := json.Marshal(&pm); err == nil {
			metadata.AddMeta("", ts, "psMeta", string(j), true)
		} else {
			slog.Error(err)
		}
	}, "chassis", "pwrsupplies")
	return md, nil
}

func c_diskspace_windows() (opentsdb.MultiDataPoint, error) {
	var dst []Win32_LogicalDisk
	var q = wmi.CreateQuery(&dst, "WHERE DriveType = 3 AND FreeSpace <> null")
	err := queryWmi(q, &dst)
	if err != nil {
		return nil, err
	}
	var md opentsdb.MultiDataPoint
	for _, v := range dst {
		tags := opentsdb.TagSet{"disk": v.Name}
		space_used := v.Size - v.FreeSpace
		Add(&md, "win.disk.fs.space_free", v.FreeSpace, tags, metadata.Gauge, metadata.Bytes, osDiskFreeDesc)
		Add(&md, "win.disk.fs.space_total", v.Size, tags, metadata.Gauge, metadata.Bytes, osDiskTotalDesc)
		Add(&md, "win.disk.fs.space_used", space_used, tags, metadata.Gauge, metadata.Bytes, osDiskUsedDesc)
		Add(&md, osDiskFree, v.FreeSpace, tags, metadata.Gauge, metadata.Bytes, osDiskFreeDesc)
		Add(&md, osDiskTotal, v.Size, tags, metadata.Gauge, metadata.Bytes, osDiskTotalDesc)
		Add(&md, osDiskUsed, space_used, tags, metadata.Gauge, metadata.Bytes, osDiskUsedDesc)
		if v.Size != 0 {
			percent_free := float64(v.FreeSpace) / float64(v.Size) * 100
			Add(&md, "win.disk.fs.percent_free", percent_free, tags, metadata.Gauge, metadata.Pct, osDiskPctFreeDesc)
			Add(&md, osDiskPctFree, percent_free, tags, metadata.Gauge, metadata.Pct, osDiskPctFreeDesc)
		}
		if v.VolumeName != "" {
			metadata.AddMeta("", tags, "label", v.VolumeName, true)
		}
	}
	return md, nil
}

func c_meta_linux_serial() (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	_ = util.ReadCommand(func(line string) error {
		fields := strings.SplitN(line, ":", 2)
		if len(fields) != 2 {
			return nil
		}
		switch fields[0] {
		case "\tSerial Number":
			metadata.AddMeta("", nil, "serialNumber", strings.TrimSpace(fields[1]), true)
		case "\tProduct Name":
			metadata.AddMeta("", nil, "model", strings.TrimSpace(fields[1]), true)
		}
		return nil
	}, "dmidecode", "-t", "system")
	return md, nil
}

func c_meta_linux_ifaces() (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	metaIfaces(func(iface net.Interface, tags opentsdb.TagSet) {
		if speed, err := ioutil.ReadFile("/sys/class/net/" + iface.Name + "/speed"); err == nil {
			v, _ := strconv.Atoi(strings.TrimSpace(string(speed)))
			if v > 0 {
				const MbitToBit = 1e6
				metadata.AddMeta("", tags, "speed", v*MbitToBit, true)
			}
		}
		_ = util.ReadCommand(func(line string) error {
			if v := metaLinuxIfacesMaster(line); v != "" {
				metadata.AddMeta("", tags, "master", v, true)
				return doneErr
			}
			return nil
		}, "ip", "-o", "addr", "show", iface.Name)
	})
	return md, nil
}

func c_fortinet_meta(host, community string) (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	ts := opentsdb.TagSet{"host": host}
	serial, err := snmpOidString(host, community, fortinetBaseOID+fortinetSerial)
	if err != nil {
		return md, fmt.Errorf("failed to get serial for host %v: %v", host, err)
	}
	metadata.AddMeta("", ts, "serialNumber", serial, false)
	version, err := snmpOidString(host, community, fortinetBaseOID+fortinetVersion)
	if err != nil {
		return md, fmt.Errorf("failed to get serial for host %v: %v", host, err)
	}
	if version == "" {
		return md, fmt.Errorf("got empty os version string for host %v", host)
	}
	// Fortinet could come from the manufactor oid, but since this is a fortinet
	// only collector saving the extra poll call
	metadata.AddMeta("", ts, "versionCaption", fmt.Sprintf("Fortinet: %v", version), false)
	return md, nil
}

func c_meta_linux_version() (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	_ = util.ReadCommand(func(line string) error {
		metadata.AddMeta("", nil, "uname", line, true)
		return nil
	}, "uname", "-a")
	if !readOSRelease() {
		readIssue()
	}
	return md, nil
}

func c_cisco_desc(host, community string) (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	desc, err := getSNMPDesc(host, community)
	if err != nil {
		return md, err
	}
	if desc == "" {
		return md, fmt.Errorf("empty description string (used to get OS version) for cisco host %v", host)
	}
	metadata.AddMeta("", opentsdb.TagSet{"host": host}, "versionCaption", desc, false)
	return md, nil
}

func metaIfaces(f func(iface net.Interface, tags opentsdb.TagSet)) {
	ifaces, _ := net.Interfaces()
	for _, iface := range ifaces {
		if strings.HasPrefix(iface.Name, "lo") {
			continue
		}
		tags := opentsdb.TagSet{"iface": fmt.Sprint("Interface", iface.Index)}
		metadata.AddMeta("", tags, "name", iface.Name, true)
		if mac := iface.HardwareAddr.String(); mac != "" {
			metadata.AddMeta("", tags, "mac", iface.HardwareAddr.String(), true)
		}
		ads, _ := iface.Addrs()
		for i, ad := range ads {
			addr := strings.Split(ad.String(), "/")[0]
			metadata.AddMeta("", opentsdb.TagSet{"addr": fmt.Sprint("Addr", i)}.Merge(tags), "addr", addr, true)
		}
		if f != nil {
			f(iface, tags)
		}
	}
}

func putMetadata(w http.ResponseWriter, r *http.Request) {
	defer r.Body.Close()
	d := json.NewDecoder(r.Body)
	var ms []metadata.Metasend
	if err := d.Decode(&ms); err != nil {
		w.WriteHeader(500)
		return
	}
	for _, m := range ms {
		metadata.AddMeta(m.Metric, m.Tags.Copy(), m.Name, m.Value, true)
	}
	w.WriteHeader(204)
}

// c_omreport_storage_pdisk is called from the controller func, since it needs the encapsulating id.
func c_omreport_storage_pdisk(id string, md *opentsdb.MultiDataPoint) {
	readOmreport(func(fields []string) {
		if len(fields) < 3 || fields[0] == "ID" {
			return
		}
		//Need to find out what the various ID formats might be
		id := strings.Replace(fields[0], ":", "_", -1)
		ts := opentsdb.TagSet{"id": id}
		Add(md, "hw.storage.pdisk", severity(fields[1]), ts, metadata.Gauge, metadata.Ok, descDellHWPDisk)
		if len(fields) < 32 {
			return
		}
		dm := &metadata.HWDiskMeta{}
		if fields[2] != "" {
			dm.Name = fields[2]
		}
		if fields[6] != "" {
			dm.Media = fields[6]
		}
		if fields[19] != "" {
			dm.Capacity = fields[19]
		}
		if fields[23] != "" {
			dm.VendorId = fields[23]
		}
		if fields[24] != "" {
			dm.ProductId = fields[24]
		}
		if fields[25] != "" {
			dm.Serial = fields[25]
		}
		if fields[26] != "" {
			dm.Part = fields[26]
		}
		if fields[27] != "" {
			dm.NegotatiedSpeed = fields[27]
		}
		if fields[28] != "" {
			dm.CapableSpeed = fields[28]
		}
		if fields[31] != "" {
			dm.SectorSize = fields[31]

		}
		if j, err := json.Marshal(&dm); err == nil {
			metadata.AddMeta("", ts, "physicalDiskMeta", string(j), true)
		} else {
			slog.Error(err)
		}
	}, "storage", "pdisk", "controller="+id)
}

func puppet_linux() (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	// See if puppet has been disabled (i.e. `puppet agent --disable 'Reason'`)
	var disabled, noReason int
	if v, err := ioutil.ReadFile(puppetDisabled); err == nil {
		disabled = 1
		d := struct {
			Disabled string `json:"disabled_message"`
		}{}
		if err := json.Unmarshal(v, &d); err == nil && d.Disabled != "" {
			if d.Disabled == "reason not specified" {
				noReason = 1
			}
			metadata.AddMeta("", nil, "puppet.disabled_reason", d.Disabled, true)
		}
	}
	Add(&md, "puppet.disabled", disabled, nil, metadata.Gauge, metadata.Count, "")
	Add(&md, "puppet.disabled_no_reason", noReason, nil, metadata.Gauge, metadata.Count, "")
	// Gather stats from the run summary
	s, err := ioutil.ReadFile(puppetRunSummary)
	if err != nil {
		return nil, err
	}
	var m PRSummary
	if err = yaml.Unmarshal(s, &m); err != nil {
		return nil, err
	}
	last_run, err := strconv.ParseInt(m.Time["last_run"], 10, 64)
	//m.Version.Config appears to be the unix timestamp
	AddTS(&md, "puppet.run.resources", last_run, m.Resources.Changed, opentsdb.TagSet{"resource": "changed"}, metadata.Gauge, metadata.Count, descPuppetChanged)
	AddTS(&md, "puppet.run.resources", last_run, m.Resources.Failed, opentsdb.TagSet{"resource": "failed"}, metadata.Gauge, metadata.Count, descPuppetFailed)
	AddTS(&md, "puppet.run.resources", last_run, m.Resources.FailedToRestart, opentsdb.TagSet{"resource": "failed_to_restart"}, metadata.Gauge, metadata.Count, descPuppetFailedToRestart)
	AddTS(&md, "puppet.run.resources", last_run, m.Resources.OutOfSync, opentsdb.TagSet{"resource": "out_of_sync"}, metadata.Gauge, metadata.Count, descPuppetOutOfSync)
	AddTS(&md, "puppet.run.resources", last_run, m.Resources.Restarted, opentsdb.TagSet{"resource": "restarted"}, metadata.Gauge, metadata.Count, descPuppetRestarted)
	AddTS(&md, "puppet.run.resources", last_run, m.Resources.Scheduled, opentsdb.TagSet{"resource": "scheduled"}, metadata.Gauge, metadata.Count, descPuppetScheduled)
	AddTS(&md, "puppet.run.resources", last_run, m.Resources.Skipped, opentsdb.TagSet{"resource": "skipped"}, metadata.Gauge, metadata.Count, descPuppetSkipped)
	AddTS(&md, "puppet.run.resources_total", last_run, m.Resources.Total, nil, metadata.Gauge, metadata.Count, descPuppetTotalResources)
	AddTS(&md, "puppet.run.changes", last_run, m.Changes.Total, nil, metadata.Gauge, metadata.Count, descPuppetTotalChanges)
	for k, v := range m.Time {
		metric, err := strconv.ParseFloat(v, 64)
		if err != nil {
			if k == "total" {
				AddTS(&md, "puppet.run_duration_total", last_run, metric, nil, metadata.Gauge, metadata.Second, descPuppetTotalTime)
			} else {
				AddTS(&md, "puppet.run_duration", last_run, metric, opentsdb.TagSet{"time": k}, metadata.Gauge, metadata.Second, descPuppetModuleTime)
			}
		}
	}
	return md, nil
}

func c_snmp_ips(community, host string) (opentsdb.MultiDataPoint, error) {
	ifIPAdEntAddrRaw, err := snmp_subtree(host, community, ifIPAdEntAddr)
	if err != nil {
		return nil, err
	}
	ipAdEnts := make(map[string]*ipAdEntAddr)
	for id, value := range ifIPAdEntAddrRaw {
		// Split entry type id from ip address
		sp := strings.SplitN(id, ".", 2)
		if len(sp) != 2 {
			slog.Errorln("unexpected length of snmp resonse")
		}
		typeId := sp[0]
		address := sp[1]
		if _, ok := ipAdEnts[address]; !ok {
			ipAdEnts[address] = &ipAdEntAddr{}
		}
		switch typeId {
		case "1":
			if v, ok := value.([]byte); ok {
				ipAdEnts[address].IP = v
			}
		case "2":
			if v, ok := value.(int64); ok {
				ipAdEnts[address].InterfaceId = v
			}
		case "3":
			if v, ok := value.([]byte); ok {
				ipAdEnts[address].Mask = v
			}
		}
	}
	ipsByInt := make(map[int64][]net.IPNet)
	for _, ipNet := range ipAdEnts {
		ipsByInt[ipNet.InterfaceId] = append(ipsByInt[ipNet.InterfaceId], ipNet.IPNet)
	}
	for intId, ipNets := range ipsByInt {
		var ips []string
		for _, ipNet := range ipNets {
			ips = append(ips, ipNet.String())
		}
		sort.Strings(ips)
		j, err := json.Marshal(ips)
		if err != nil {
			slog.Errorf("error marshaling ips for host %v: %v", host, err)
		}
		metadata.AddMeta("", opentsdb.TagSet{"host": host, "iface": fmt.Sprintf("%v", intId)}, "addresses", string(j), false)
	}
	return nil, nil
}

func metaIfaces(f func(iface net.Interface, tags opentsdb.TagSet)) {
	ifaces, _ := net.Interfaces()
	for _, iface := range ifaces {
		if strings.HasPrefix(iface.Name, "lo") {
			continue
		}
		tags := opentsdb.TagSet{"iface": iface.Name}
		metadata.AddMeta("", tags, "name", iface.Name, true)
		if mac := strings.ToUpper(strings.Replace(iface.HardwareAddr.String(), ":", "", -1)); mac != "" {
			metadata.AddMeta("", tags, "mac", mac, true)
		}
		rawAds, _ := iface.Addrs()
		addrs := make([]string, len(rawAds))
		for i, rAd := range rawAds {
			addrs[i] = rAd.String()
		}
		sort.Strings(addrs)
		j, _ := json.Marshal(addrs)
		metadata.AddMeta("", tags, "addresses", string(j), true)
		if f != nil {
			f(iface, tags)
		}
	}
}

func c_omreport_memory() (opentsdb.MultiDataPoint, error) {
	var md opentsdb.MultiDataPoint
	readOmreport(func(fields []string) {
		if len(fields) != 5 {
			return
		}
		if _, err := strconv.Atoi(fields[0]); err != nil {
			return
		}
		ts := opentsdb.TagSet{"name": replace(fields[2])}
		Add(&md, "hw.chassis.memory", severity(fields[1]), ts, metadata.Gauge, metadata.Ok, descDellHWMemory)
		metadata.AddMeta("", ts, "memory", clean(fields[4]), true)
	}, "chassis", "memory")
	return md, nil
}

func readOSRelease() bool {
	var found bool
	_ = readLine("/etc/os-release", func(s string) error {
		fields := strings.SplitN(s, "=", 2)
		if len(fields) != 2 {
			return nil
		}
		if fields[0] == "PRETTY_NAME" {
			metadata.AddMeta("", nil, "version", strings.Trim(fields[1], `"`), true)
			found = true
		}
		return nil
	})
	return found
}