telegraf/plugins/inputs/rss/rss.go

96 lines
1.6 KiB
Go

package rss
import (
"fmt"
"sync"
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/inputs"
"github.com/mmcdole/gofeed"
)
var ts time.Time
type RSS struct {
URLs []string `toml:"urls"`
}
func (r *RSS) Description() string {
return "Reads RSS Feeds Description"
}
var sampleConfig = `
## One or more URLs from which to read rss feed
urls = [
"http://localhost/rss/feed"
]
## Reduce traffic. Do not run too frequently.
interval = '10m'
`
func (r *RSS) SampleConfig() string {
return sampleConfig
}
func (r *RSS) Gather(acc telegraf.Accumulator) error {
if ts.IsZero() {
ts = time.Now()
}
var wg sync.WaitGroup
for _, u := range r.URLs {
wg.Add(1)
go func(url string) {
defer wg.Done()
if err := r.gatherFeed(acc, url, ts); err != nil {
acc.AddError(fmt.Errorf("[url=%s]: %s", url, err))
}
}(u)
}
// remember last time started
tsl := time.Now()
wg.Wait()
// set last time started
ts = tsl
return nil
}
func (r *RSS) gatherFeed(acc telegraf.Accumulator, url string, ts time.Time) error {
fp := gofeed.NewParser()
feed, err := fp.ParseURL(url)
if err != nil {
return err
}
fields := make(map[string]interface{})
tags := make(map[string]string)
tags["url"] = url
tags["feed_type"] = feed.FeedType
tags["feed_version"] = feed.FeedVersion
for _, item := range feed.Items {
pubtime := item.PublishedParsed
if pubtime.Before(ts) {
continue
}
fields["title"] = item.Title
fields["description"] = item.Description
acc.AddFields("rss", fields, tags, *pubtime)
}
return nil
}
func init() {
inputs.Add("rss", func() telegraf.Input { return &RSS{} })
}