package rabbitmq import ( "encoding/json" "errors" "fmt" "io" "log" "net/http" "sync" "sync/atomic" "time" "flashcat.cloud/categraf/config" "flashcat.cloud/categraf/inputs" "flashcat.cloud/categraf/pkg/filter" "flashcat.cloud/categraf/pkg/tls" "flashcat.cloud/categraf/types" "github.com/toolkits/pkg/container/list" ) const inputName = "rabbitmq" type RabbitMQ struct { config.Interval counter uint64 waitgrp sync.WaitGroup Instances []*Instance `toml:"instances"` } func init() { inputs.Add(inputName, func() inputs.Input { return &RabbitMQ{} }) } func (r *RabbitMQ) Prefix() string { return inputName } func (r *RabbitMQ) Init() error { if len(r.Instances) == 0 { return types.ErrInstancesEmpty } for i := 0; i < len(r.Instances); i++ { if err := r.Instances[i].Init(); err != nil { return err } } return nil } func (r *RabbitMQ) Drop() {} func (r *RabbitMQ) Gather(slist *list.SafeList) { atomic.AddUint64(&r.counter, 1) for i := range r.Instances { ins := r.Instances[i] r.waitgrp.Add(1) go func(slist *list.SafeList, ins *Instance) { defer r.waitgrp.Done() if ins.IntervalTimes > 0 { counter := atomic.LoadUint64(&r.counter) if counter%uint64(ins.IntervalTimes) != 0 { return } } ins.gatherOnce(slist) }(slist, ins) } r.waitgrp.Wait() } type Instance struct { URL string `toml:"url"` Username string `toml:"username"` Password string `toml:"password"` Labels map[string]string `toml:"labels"` IntervalTimes int64 `toml:"interval_times"` HeaderTimeout config.Duration `toml:"header_timeout"` ClientTimeout config.Duration `toml:"client_timeout"` Nodes []string `toml:"nodes"` Exchanges []string `toml:"exchanges"` MetricInclude []string `toml:"metric_include"` MetricExclude []string `toml:"metric_exclude"` QueueInclude []string `toml:"queue_name_include"` QueueExclude []string `toml:"queue_name_exclude"` FederationUpstreamInclude []string `toml:"federation_upstream_include"` FederationUpstreamExclude []string `toml:"federation_upstream_exclude"` tls.ClientConfig client *http.Client metricFilter filter.Filter queueFilter filter.Filter upstreamFilter filter.Filter excludeEveryQueue bool } func (ins *Instance) Init() error { if ins.URL == "" { return errors.New("url is blank") } var err error if err := ins.createQueueFilter(); err != nil { return err } if ins.upstreamFilter, err = filter.NewIncludeExcludeFilter(ins.FederationUpstreamInclude, ins.FederationUpstreamExclude); err != nil { return err } if ins.metricFilter, err = filter.NewIncludeExcludeFilter(ins.MetricInclude, ins.MetricExclude); err != nil { return err } ins.client, err = ins.createHTTPClient() return err } func (ins *Instance) createQueueFilter() error { queueFilter, err := filter.NewIncludeExcludeFilter(ins.QueueInclude, ins.QueueExclude) if err != nil { return err } ins.queueFilter = queueFilter for _, q := range ins.QueueExclude { if q == "*" { ins.excludeEveryQueue = true } } return nil } func (ins *Instance) createHTTPClient() (*http.Client, error) { if ins.HeaderTimeout <= 0 { ins.HeaderTimeout = config.Duration(time.Second * 3) } if ins.ClientTimeout <= 0 { ins.ClientTimeout = config.Duration(time.Second * 4) } trans := &http.Transport{ ResponseHeaderTimeout: time.Duration(ins.HeaderTimeout), } if ins.UseTLS { tlsConfig, err := ins.ClientConfig.TLSConfig() if err != nil { return nil, err } trans.TLSClientConfig = tlsConfig } client := &http.Client{ Transport: trans, Timeout: time.Duration(ins.ClientTimeout), } return client, nil } // OverviewResponse ... type OverviewResponse struct { MessageStats *MessageStats `json:"message_stats"` ObjectTotals *ObjectTotals `json:"object_totals"` QueueTotals *QueueTotals `json:"queue_totals"` Listeners []Listeners `json:"listeners"` } // Listeners ... type Listeners struct { Protocol string `json:"protocol"` } // Details ... type Details struct { Rate float64 `json:"rate"` } // MessageStats ... type MessageStats struct { Ack int64 AckDetails Details `json:"ack_details"` Deliver int64 DeliverDetails Details `json:"deliver_details"` DeliverGet int64 `json:"deliver_get"` DeliverGetDetails Details `json:"deliver_get_details"` Publish int64 PublishDetails Details `json:"publish_details"` Redeliver int64 RedeliverDetails Details `json:"redeliver_details"` PublishIn int64 `json:"publish_in"` PublishInDetails Details `json:"publish_in_details"` PublishOut int64 `json:"publish_out"` PublishOutDetails Details `json:"publish_out_details"` ReturnUnroutable int64 `json:"return_unroutable"` ReturnUnroutableDetails Details `json:"return_unroutable_details"` } // ObjectTotals ... type ObjectTotals struct { Channels int64 Connections int64 Consumers int64 Exchanges int64 Queues int64 } // QueueTotals ... type QueueTotals struct { Messages int64 MessagesReady int64 `json:"messages_ready"` MessagesUnacknowledged int64 `json:"messages_unacknowledged"` MessageBytes int64 `json:"message_bytes"` MessageBytesReady int64 `json:"message_bytes_ready"` MessageBytesUnacknowledged int64 `json:"message_bytes_unacknowledged"` MessageRAM int64 `json:"message_bytes_ram"` MessagePersistent int64 `json:"message_bytes_persistent"` } // Queue ... type Queue struct { QueueTotals // just to not repeat the same code MessageStats `json:"message_stats"` Memory int64 Consumers int64 ConsumerUtilisation float64 `json:"consumer_utilisation"` Name string Node string Vhost string Durable bool AutoDelete bool `json:"auto_delete"` IdleSince string `json:"idle_since"` SlaveNodes []string `json:"slave_nodes"` SynchronisedSlaveNodes []string `json:"synchronised_slave_nodes"` } // Node ... type Node struct { Name string DiskFree int64 `json:"disk_free"` DiskFreeLimit int64 `json:"disk_free_limit"` DiskFreeAlarm bool `json:"disk_free_alarm"` FdTotal int64 `json:"fd_total"` FdUsed int64 `json:"fd_used"` MemLimit int64 `json:"mem_limit"` MemUsed int64 `json:"mem_used"` MemAlarm bool `json:"mem_alarm"` ProcTotal int64 `json:"proc_total"` ProcUsed int64 `json:"proc_used"` RunQueue int64 `json:"run_queue"` SocketsTotal int64 `json:"sockets_total"` SocketsUsed int64 `json:"sockets_used"` Running bool `json:"running"` Uptime int64 `json:"uptime"` MnesiaDiskTxCount int64 `json:"mnesia_disk_tx_count"` MnesiaDiskTxCountDetails Details `json:"mnesia_disk_tx_count_details"` MnesiaRAMTxCount int64 `json:"mnesia_ram_tx_count"` MnesiaRAMTxCountDetails Details `json:"mnesia_ram_tx_count_details"` GcNum int64 `json:"gc_num"` GcNumDetails Details `json:"gc_num_details"` GcBytesReclaimed int64 `json:"gc_bytes_reclaimed"` GcBytesReclaimedDetails Details `json:"gc_bytes_reclaimed_details"` IoReadAvgTime float64 `json:"io_read_avg_time"` IoReadAvgTimeDetails Details `json:"io_read_avg_time_details"` IoReadBytes int64 `json:"io_read_bytes"` IoReadBytesDetails Details `json:"io_read_bytes_details"` IoWriteAvgTime float64 `json:"io_write_avg_time"` IoWriteAvgTimeDetails Details `json:"io_write_avg_time_details"` IoWriteBytes int64 `json:"io_write_bytes"` IoWriteBytesDetails Details `json:"io_write_bytes_details"` } type Exchange struct { Name string MessageStats `json:"message_stats"` Type string Internal bool Vhost string Durable bool AutoDelete bool `json:"auto_delete"` } // FederationLinkChannelMessageStats ... type FederationLinkChannelMessageStats struct { Confirm int64 `json:"confirm"` ConfirmDetails Details `json:"confirm_details"` Publish int64 `json:"publish"` PublishDetails Details `json:"publish_details"` ReturnUnroutable int64 `json:"return_unroutable"` ReturnUnroutableDetails Details `json:"return_unroutable_details"` } // FederationLinkChannel ... type FederationLinkChannel struct { AcksUncommitted int64 `json:"acks_uncommitted"` ConsumerCount int64 `json:"consumer_count"` MessagesUnacknowledged int64 `json:"messages_unacknowledged"` MessagesUncommitted int64 `json:"messages_uncommitted"` MessagesUnconfirmed int64 `json:"messages_unconfirmed"` MessageStats FederationLinkChannelMessageStats `json:"message_stats"` } // FederationLink ... type FederationLink struct { Type string `json:"type"` Queue string `json:"queue"` UpstreamQueue string `json:"upstream_queue"` Exchange string `json:"exchange"` UpstreamExchange string `json:"upstream_exchange"` Vhost string `json:"vhost"` Upstream string `json:"upstream"` LocalChannel FederationLinkChannel `json:"local_channel"` } type HealthCheck struct { Status string `json:"status"` } // MemoryResponse ... type MemoryResponse struct { Memory *Memory `json:"memory"` } // Memory details type Memory struct { ConnectionReaders int64 `json:"connection_readers"` ConnectionWriters int64 `json:"connection_writers"` ConnectionChannels int64 `json:"connection_channels"` ConnectionOther int64 `json:"connection_other"` QueueProcs int64 `json:"queue_procs"` QueueSlaveProcs int64 `json:"queue_slave_procs"` Plugins int64 `json:"plugins"` OtherProc int64 `json:"other_proc"` Metrics int64 `json:"metrics"` MgmtDb int64 `json:"mgmt_db"` Mnesia int64 `json:"mnesia"` OtherEts int64 `json:"other_ets"` Binary int64 `json:"binary"` MsgIndex int64 `json:"msg_index"` Code int64 `json:"code"` Atom int64 `json:"atom"` OtherSystem int64 `json:"other_system"` AllocatedUnused int64 `json:"allocated_unused"` ReservedUnallocated int64 `json:"reserved_unallocated"` Total interface{} `json:"total"` } // Error response type ErrorResponse struct { Error string `json:"error"` Reason string `json:"reason"` } // gatherFunc ... type gatherFunc func(ins *Instance, slist *list.SafeList) var gatherFunctions = map[string]gatherFunc{ "exchange": gatherExchanges, "federation": gatherFederationLinks, "node": gatherNodes, "overview": gatherOverview, "queue": gatherQueues, } func (ins *Instance) gatherOnce(slist *list.SafeList) { tags := map[string]string{"url": ins.URL} begun := time.Now() // scrape use seconds defer func(begun time.Time) { use := time.Since(begun).Seconds() slist.PushFront(inputs.NewSample("scrape_use_seconds", use, tags, ins.Labels)) }(begun) var wg sync.WaitGroup for name, f := range gatherFunctions { // Query only metrics that are supported if !ins.metricFilter.Match(name) { continue } wg.Add(1) go func(gf gatherFunc) { defer wg.Done() gf(ins, slist) }(f) } wg.Wait() } func (ins *Instance) requestEndpoint(u string) ([]byte, error) { endpoint := ins.URL + u if config.Config.DebugMode { log.Println("D! requesting:", endpoint) } req, err := http.NewRequest("GET", endpoint, nil) if err != nil { return nil, err } req.SetBasicAuth(ins.Username, ins.Password) resp, err := ins.client.Do(req) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode/100 != 2 { return nil, fmt.Errorf("getting %q failed: %v %v", u, resp.StatusCode, http.StatusText(resp.StatusCode)) } return io.ReadAll(resp.Body) } func (ins *Instance) requestJSON(u string, target interface{}) error { buf, err := ins.requestEndpoint(u) if err != nil { return err } if err := json.Unmarshal(buf, target); err != nil { if _, ok := err.(*json.UnmarshalTypeError); ok { // Try to get the error reason from the response var errResponse ErrorResponse if json.Unmarshal(buf, &errResponse) == nil && errResponse.Error != "" { // Return the error reason in the response return fmt.Errorf("error response trying to get %q: %q (reason: %q)", u, errResponse.Error, errResponse.Reason) } } return fmt.Errorf("decoding answer from %q failed: %v", u, err) } return nil } func gatherOverview(ins *Instance, slist *list.SafeList) { overview := OverviewResponse{} err := ins.requestJSON("/api/overview", &overview) if err != nil { log.Println("E! failed to query rabbitmq /api/overview:", err) return } if overview.QueueTotals == nil || overview.ObjectTotals == nil || overview.MessageStats == nil || overview.Listeners == nil { log.Println("E! wrong answer from rabbitmq. probably auth issue") return } var clusteringListeners, amqpListeners int64 = 0, 0 for _, listener := range overview.Listeners { if listener.Protocol == "clustering" { clusteringListeners++ } else if listener.Protocol == "amqp" { amqpListeners++ } } tags := map[string]string{"url": ins.URL} for k, v := range ins.Labels { tags[k] = v } fields := map[string]interface{}{ "overview_messages": overview.QueueTotals.Messages, "overview_messages_ready": overview.QueueTotals.MessagesReady, "overview_messages_unacked": overview.QueueTotals.MessagesUnacknowledged, "overview_channels": overview.ObjectTotals.Channels, "overview_connections": overview.ObjectTotals.Connections, "overview_consumers": overview.ObjectTotals.Consumers, "overview_exchanges": overview.ObjectTotals.Exchanges, "overview_queues": overview.ObjectTotals.Queues, "overview_messages_acked": overview.MessageStats.Ack, "overview_messages_acked_rate": overview.MessageStats.AckDetails.Rate, "overview_messages_delivered": overview.MessageStats.Deliver, "overview_messages_delivered_rate": overview.MessageStats.DeliverDetails.Rate, "overview_messages_redelivered": overview.MessageStats.Redeliver, "overview_messages_redelivered_rate": overview.MessageStats.RedeliverDetails.Rate, "overview_messages_delivered_get": overview.MessageStats.DeliverGet, "overview_messages_published": overview.MessageStats.Publish, "overview_clustering_listeners": clusteringListeners, "overview_amqp_listeners": amqpListeners, "overview_return_unroutable": overview.MessageStats.ReturnUnroutable, "overview_return_unroutable_rate": overview.MessageStats.ReturnUnroutableDetails.Rate, } inputs.PushSamples(slist, fields, tags) } func gatherExchanges(ins *Instance, slist *list.SafeList) { // Gather information about exchanges exchanges := make([]Exchange, 0) err := ins.requestJSON("/api/exchanges", &exchanges) if err != nil { log.Println("E! failed to query rabbitmq /api/exchanges:", err) return } for _, exchange := range exchanges { if !ins.shouldGatherExchange(exchange.Name) { continue } tags := map[string]string{ "url": ins.URL, "exchange": exchange.Name, "type": exchange.Type, "vhost": exchange.Vhost, // "internal": strconv.FormatBool(exchange.Internal), // "durable": strconv.FormatBool(exchange.Durable), // "auto_delete": strconv.FormatBool(exchange.AutoDelete), } for k, v := range ins.Labels { tags[k] = v } fields := map[string]interface{}{ "exchange_messages_publish_in": exchange.MessageStats.PublishIn, "exchange_messages_publish_in_rate": exchange.MessageStats.PublishInDetails.Rate, "exchange_messages_publish_out": exchange.MessageStats.PublishOut, "exchange_messages_publish_out_rate": exchange.MessageStats.PublishOutDetails.Rate, } inputs.PushSamples(slist, fields, tags) } } func (ins *Instance) shouldGatherExchange(exchangeName string) bool { if len(ins.Exchanges) == 0 { return true } for _, name := range ins.Exchanges { if name == exchangeName { return true } } return false } func gatherFederationLinks(ins *Instance, slist *list.SafeList) { // Gather information about federation links federationLinks := make([]FederationLink, 0) err := ins.requestJSON("/api/federation-links", &federationLinks) if err != nil { log.Println("E! failed to query rabbitmq /api/federation-links:", err) return } for _, link := range federationLinks { if !ins.shouldGatherFederationLink(link) { continue } tags := map[string]string{ "url": ins.URL, "type": link.Type, "vhost": link.Vhost, "upstream": link.Upstream, } if link.Type == "exchange" { tags["exchange"] = link.Exchange tags["upstream_exchange"] = link.UpstreamExchange } else { tags["queue"] = link.Queue tags["upstream_queue"] = link.UpstreamQueue } fields := map[string]interface{}{ "federation_acks_uncommitted": link.LocalChannel.AcksUncommitted, "federation_consumers": link.LocalChannel.ConsumerCount, "federation_messages_unacknowledged": link.LocalChannel.MessagesUnacknowledged, "federation_messages_uncommitted": link.LocalChannel.MessagesUncommitted, "federation_messages_unconfirmed": link.LocalChannel.MessagesUnconfirmed, "federation_messages_confirm": link.LocalChannel.MessageStats.Confirm, "federation_messages_publish": link.LocalChannel.MessageStats.Publish, "federation_messages_return_unroutable": link.LocalChannel.MessageStats.ReturnUnroutable, } inputs.PushSamples(slist, fields, tags, ins.Labels) } } func (ins *Instance) shouldGatherFederationLink(link FederationLink) bool { if !ins.upstreamFilter.Match(link.Upstream) { return false } switch link.Type { case "exchange": return ins.shouldGatherExchange(link.Exchange) case "queue": return ins.queueFilter.Match(link.Queue) default: return false } } func gatherNodes(ins *Instance, slist *list.SafeList) { allNodes := make([]*Node, 0) err := ins.requestJSON("/api/nodes", &allNodes) if err != nil { log.Println("E! failed to query rabbitmq /api/nodes:", err) return } nodes := allNodes[:0] for _, node := range allNodes { if ins.shouldGatherNode(node) { nodes = append(nodes, node) } } var wg sync.WaitGroup for _, node := range nodes { wg.Add(1) go func(node *Node) { defer wg.Done() tags := map[string]string{"url": ins.URL} tags["node"] = node.Name fields := map[string]interface{}{ "node_disk_free": node.DiskFree, "node_disk_free_limit": node.DiskFreeLimit, "node_disk_free_alarm": boolToInt(node.DiskFreeAlarm), "node_fd_total": node.FdTotal, "node_fd_used": node.FdUsed, "node_mem_limit": node.MemLimit, "node_mem_used": node.MemUsed, "node_mem_alarm": boolToInt(node.MemAlarm), "node_proc_total": node.ProcTotal, "node_proc_used": node.ProcUsed, "node_run_queue": node.RunQueue, "node_sockets_total": node.SocketsTotal, "node_sockets_used": node.SocketsUsed, "node_uptime": node.Uptime, "node_mnesia_disk_tx_count": node.MnesiaDiskTxCount, "node_mnesia_disk_tx_count_rate": node.MnesiaDiskTxCountDetails.Rate, "node_mnesia_ram_tx_count": node.MnesiaRAMTxCount, "node_mnesia_ram_tx_count_rate": node.MnesiaRAMTxCountDetails.Rate, "node_gc_num": node.GcNum, "node_gc_num_rate": node.GcNumDetails.Rate, "node_gc_bytes_reclaimed": node.GcBytesReclaimed, "node_gc_bytes_reclaimed_rate": node.GcBytesReclaimedDetails.Rate, "node_io_read_avg_time": node.IoReadAvgTime, "node_io_read_avg_time_rate": node.IoReadAvgTimeDetails.Rate, "node_io_read_bytes": node.IoReadBytes, "node_io_read_bytes_rate": node.IoReadBytesDetails.Rate, "node_io_write_avg_time": node.IoWriteAvgTime, "node_io_write_avg_time_rate": node.IoWriteAvgTimeDetails.Rate, "node_io_write_bytes": node.IoWriteBytes, "node_io_write_bytes_rate": node.IoWriteBytesDetails.Rate, "node_running": boolToInt(node.Running), } var memory MemoryResponse err = ins.requestJSON("/api/nodes/"+node.Name+"/memory", &memory) if err != nil { log.Println("E! failed to query rabbitmq /api/nodes/"+node.Name+"/memory:", err) return } if memory.Memory != nil { fields["node_mem_connection_readers"] = memory.Memory.ConnectionReaders fields["node_mem_connection_writers"] = memory.Memory.ConnectionWriters fields["node_mem_connection_channels"] = memory.Memory.ConnectionChannels fields["node_mem_connection_other"] = memory.Memory.ConnectionOther fields["node_mem_queue_procs"] = memory.Memory.QueueProcs fields["node_mem_queue_slave_procs"] = memory.Memory.QueueSlaveProcs fields["node_mem_plugins"] = memory.Memory.Plugins fields["node_mem_other_proc"] = memory.Memory.OtherProc fields["node_mem_metrics"] = memory.Memory.Metrics fields["node_mem_mgmt_db"] = memory.Memory.MgmtDb fields["node_mem_mnesia"] = memory.Memory.Mnesia fields["node_mem_other_ets"] = memory.Memory.OtherEts fields["node_mem_binary"] = memory.Memory.Binary fields["node_mem_msg_index"] = memory.Memory.MsgIndex fields["node_mem_code"] = memory.Memory.Code fields["node_mem_atom"] = memory.Memory.Atom fields["node_mem_other_system"] = memory.Memory.OtherSystem fields["node_mem_allocated_unused"] = memory.Memory.AllocatedUnused fields["node_mem_reserved_unallocated"] = memory.Memory.ReservedUnallocated switch v := memory.Memory.Total.(type) { case float64: fields["node_mem_total"] = int64(v) case map[string]interface{}: var foundEstimator bool for _, estimator := range []string{"rss", "allocated", "erlang"} { if x, found := v[estimator]; found { if total, ok := x.(float64); ok { fields["node_mem_total"] = int64(total) foundEstimator = true break } msg := fmt.Sprintf("unknown type %T for %q total memory", x, estimator) log.Println("E!", msg) } } if !foundEstimator { log.Println("E! no known memory estimation in", v) } default: log.Println("E! unknown type", memory.Memory.Total, "for total memory") } } inputs.PushSamples(slist, fields, tags, ins.Labels) }(node) } wg.Wait() } func (ins *Instance) shouldGatherNode(node *Node) bool { if len(ins.Nodes) == 0 { return true } for _, name := range ins.Nodes { if name == node.Name { return true } } return false } func boolToInt(b bool) int64 { if b { return 1 } return 0 } func gatherQueues(ins *Instance, slist *list.SafeList) { if ins.excludeEveryQueue { return } // Gather information about queues queues := make([]Queue, 0) err := ins.requestJSON("/api/queues", &queues) if err != nil { log.Println("E! failed to query rabbitmq /api/queues:", err) return } for _, queue := range queues { if !ins.queueFilter.Match(queue.Name) { continue } tags := map[string]string{ "url": ins.URL, "queue": queue.Name, "vhost": queue.Vhost, "node": queue.Node, // "durable": strconv.FormatBool(queue.Durable), // "auto_delete": strconv.FormatBool(queue.AutoDelete), } fields := map[string]interface{}{ // common information "queue_consumers": queue.Consumers, "queue_consumer_utilisation": queue.ConsumerUtilisation, "queue_idle_since": queue.IdleSince, "queue_slave_nodes": len(queue.SlaveNodes), "queue_synchronised_slave_nodes": len(queue.SynchronisedSlaveNodes), "queue_memory": queue.Memory, // messages information "queue_message_bytes": queue.MessageBytes, "queue_message_bytes_ready": queue.MessageBytesReady, "queue_message_bytes_unacked": queue.MessageBytesUnacknowledged, "queue_message_bytes_ram": queue.MessageRAM, "queue_message_bytes_persist": queue.MessagePersistent, "queue_messages": queue.Messages, "queue_messages_ready": queue.MessagesReady, "queue_messages_unack": queue.MessagesUnacknowledged, "queue_messages_ack": queue.MessageStats.Ack, "queue_messages_ack_rate": queue.MessageStats.AckDetails.Rate, "queue_messages_deliver": queue.MessageStats.Deliver, "queue_messages_deliver_rate": queue.MessageStats.DeliverDetails.Rate, "queue_messages_deliver_get": queue.MessageStats.DeliverGet, "queue_messages_deliver_get_rate": queue.MessageStats.DeliverGetDetails.Rate, "queue_messages_publish": queue.MessageStats.Publish, "queue_messages_publish_rate": queue.MessageStats.PublishDetails.Rate, "queue_messages_redeliver": queue.MessageStats.Redeliver, "queue_messages_redeliver_rate": queue.MessageStats.RedeliverDetails.Rate, } inputs.PushSamples(slist, fields, tags, ins.Labels) } }