dom/telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Don't overwrite forecast points (#5930)

Browse Source
This commit is contained in:
Daniel Nelson
2019-05-31 16:22:37 -07:00
committed by GitHub
parent a0213d9c4f
commit 0ca8ea1724
6 changed files with 315 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files

250
plugins/inputs/openweathermap/openweathermap.go
View File

@@ -1,9 +1,10 @@
package openweathermap
import (
"bufio"
"encoding/json"
"fmt"
"io"
"mime"
"net/http"
"net/url"
"strconv"
@@ -16,37 +17,50 @@ import (
"github.com/influxdata/telegraf/plugins/inputs"
)
const (
// https://openweathermap.org/current#severalid
// Call for several city IDs
// The limit of locations is 20.
owmRequestSeveralCityId int = 20
defaultBaseURL = "https://api.openweathermap.org/"
defaultResponseTimeout time.Duration = time.Second * 5
defaultUnits string = "metric"
)
type OpenWeatherMap struct {
BaseUrl string
AppId string
CityId []string
AppId string `toml:"app_id"`
CityId []string `toml:"city_id"`
Fetch []string `toml:"fetch"`
BaseUrl string `toml:"base_url"`
ResponseTimeout internal.Duration `toml:"response_timeout"`
Units string `toml:"units"`
client *http.Client
ResponseTimeout internal.Duration
Fetch []string
Units string
}
// https://openweathermap.org/current#severalid
// Call for several city IDs
// The limit of locations is 20.
const owmRequestSeveralCityId int = 20
const defaultResponseTimeout time.Duration = time.Second * 5
const defaultUnits string = "metric"
var sampleConfig = `
## Root url of weather map REST API
base_url = "https://api.openweathermap.org/"
## Your personal user token from openweathermap.org
## OpenWeatherMap API key.
app_id = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
city_id = ["2988507", "2988588"]
## HTTP response timeout (default: 5s)
response_timeout = "5s"
## City ID's to collect weather data from.
city_id = ["5391959"]
## APIs to fetch; can contain "weather" or "forecast".
fetch = ["weather", "forecast"]
units = "metric"
## Limit OpenWeatherMap query interval. See calls per minute info at: https://openweathermap.org/price
## OpenWeatherMap base URL
# base_url = "https://api.openweathermap.org/"
## Timeout for HTTP response.
# response_timeout = "5s"
## Preferred unit system for temperature and wind speed. Can be one of
## "metric", "imperial", or "standard".
# units = "metric"
## Query interval; OpenWeatherMap updates their weather data every 10
## minutes.
interval = "10m"
`
@@ -69,7 +83,6 @@ func (n *OpenWeatherMap) Gather(acc telegraf.Accumulator) error {
// Create an HTTP client that is re-used for each
// collection interval
if n.client == nil {
client, err := n.createHttpClient()
if err != nil {
@@ -77,33 +90,43 @@ func (n *OpenWeatherMap) Gather(acc telegraf.Accumulator) error {
}
n.client = client
}
units := n.Units
if units == "" {
switch n.Units {
case "imperial", "standard":
break
default:
units = defaultUnits
}
for _, fetch := range n.Fetch {
if fetch == "forecast" {
var u *url.URL
var addr *url.URL
for _, city := range n.CityId {
u, err = url.Parse(fmt.Sprintf("/data/2.5/forecast?id=%s&APPID=%s&units=%s", city, n.AppId, units))
if err != nil {
acc.AddError(fmt.Errorf("Unable to parse address '%s': %s", u, err))
acc.AddError(fmt.Errorf("unable to parse address '%s': %s", u, err))
continue
}
addr = base.ResolveReference(u)
addr := base.ResolveReference(u).String()
wg.Add(1)
go func(addr *url.URL) {
go func() {
defer wg.Done()
acc.AddError(n.gatherUrl(addr, acc, true))
}(addr)
status, err := n.gatherUrl(addr)
if err != nil {
acc.AddError(err)
return
}
gatherForecast(acc, status)
}()
}
} else if fetch == "weather" {
j := 0
for j < len(n.CityId) {
var u *url.URL
var addr *url.URL
strs = make([]string, 0)
for i := 0; j < len(n.CityId) && i < owmRequestSeveralCityId; i++ {
strs = append(strs, n.CityId[j])
@@ -117,12 +140,18 @@ func (n *OpenWeatherMap) Gather(acc telegraf.Accumulator) error {
continue
}
addr = base.ResolveReference(u)
addr := base.ResolveReference(u).String()
wg.Add(1)
go func(addr *url.URL) {
go func() {
defer wg.Done()
acc.AddError(n.gatherUrl(addr, acc, false))
}(addr)
status, err := n.gatherUrl(addr)
if err != nil {
acc.AddError(err)
return
}
gatherWeather(acc, status)
}()
}
}
@@ -133,7 +162,6 @@ func (n *OpenWeatherMap) Gather(acc telegraf.Accumulator) error {
}
func (n *OpenWeatherMap) createHttpClient() (*http.Client, error) {
if n.ResponseTimeout.Duration < time.Second {
n.ResponseTimeout.Duration = defaultResponseTimeout
}
@@ -146,65 +174,51 @@ func (n *OpenWeatherMap) createHttpClient() (*http.Client, error) {
return client, nil
}
func (n *OpenWeatherMap) gatherUrl(addr *url.URL, acc telegraf.Accumulator, forecast bool) error {
resp, err := n.client.Get(addr.String())
func (n *OpenWeatherMap) gatherUrl(addr string) (*Status, error) {
resp, err := n.client.Get(addr)
if err != nil {
return fmt.Errorf("error making HTTP request to %s: %s", addr.String(), err)
return nil, fmt.Errorf("error making HTTP request to %s: %s", addr, err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("%s returned HTTP status %s", addr.String(), resp.Status)
return nil, fmt.Errorf("%s returned HTTP status %s", addr, resp.Status)
}
contentType := strings.Split(resp.Header.Get("Content-Type"), ";")[0]
switch contentType {
case "application/json":
err = gatherWeatherUrl(bufio.NewReader(resp.Body), forecast, acc)
return err
default:
return fmt.Errorf("%s returned unexpected content type %s", addr.String(), contentType)
mediaType, _, err := mime.ParseMediaType(resp.Header.Get("Content-Type"))
if err != nil {
return nil, err
}
if mediaType != "application/json" {
return nil, fmt.Errorf("%s returned unexpected content type %s", addr, mediaType)
}
return gatherWeatherUrl(resp.Body)
}
type WeatherEntry struct {
Dt int64 `json:"dt"`
Dttxt string `json:"dt_txt"` // empty for weather/
Dt int64 `json:"dt"`
Clouds struct {
All int64 `json:"all"`
} `json:"clouds"`
Main struct {
GrndLevel float64 `json:"grnd_level"` // empty for weather/
Humidity int64 `json:"humidity"`
SeaLevel float64 `json:"sea_level"` // empty for weather/
Pressure float64 `json:"pressure"`
Temp float64 `json:"temp"`
TempMax float64 `json:"temp_max"`
TempMin float64 `json:"temp_min"`
Humidity int64 `json:"humidity"`
Pressure float64 `json:"pressure"`
Temp float64 `json:"temp"`
} `json:"main"`
Rain struct {
Rain3 float64 `json:"3h"`
} `json:"rain"`
Sys struct {
Pod string `json:"pod"`
Country string `json:"country"`
Message float64 `json:"message"`
Id int64 `json:"id"`
Type int64 `json:"type"`
Sunrise int64 `json:"sunrise"`
Sunset int64 `json:"sunset"`
Country string `json:"country"`
Sunrise int64 `json:"sunrise"`
Sunset int64 `json:"sunset"`
} `json:"sys"`
Wind struct {
Deg float64 `json:"deg"`
Speed float64 `json:"speed"`
} `json:"wind"`
Weather []struct {
Description string `json:"description"`
Icon string `json:"icon"`
Id int64 `json:"id"`
Main string `json:"main"`
} `json:"weather"`
// Additional entries for weather/
Id int64 `json:"id"`
Name string `json:"name"`
Coord struct {
@@ -227,69 +241,66 @@ type Status struct {
List []WeatherEntry `json:"list"`
}
func gatherWeatherUrl(r *bufio.Reader, forecast bool, acc telegraf.Accumulator) error {
func gatherWeatherUrl(r io.Reader) (*Status, error) {
dec := json.NewDecoder(r)
status := &Status{}
if err := dec.Decode(status); err != nil {
return fmt.Errorf("Error while decoding JSON response: %s", err)
return nil, fmt.Errorf("error while decoding JSON response: %s", err)
}
status.Gather(forecast, acc)
return nil
return status, nil
}
func (s *Status) Gather(forecast bool, acc telegraf.Accumulator) {
tags := map[string]string{
"city_id": strconv.FormatInt(s.City.Id, 10),
"forecast": "*",
}
for i, e := range s.List {
func gatherWeather(acc telegraf.Accumulator, status *Status) {
for _, e := range status.List {
tm := time.Unix(e.Dt, 0)
if e.Id > 0 {
tags["city_id"] = strconv.FormatInt(e.Id, 10)
}
if forecast {
tags["forecast"] = fmt.Sprintf("%dh", (i+1)*3)
}
acc.AddFields(
"weather",
map[string]interface{}{
"rain": e.Rain.Rain3,
"wind_degrees": e.Wind.Deg,
"wind_speed": e.Wind.Speed,
"cloudiness": e.Clouds.All,
"humidity": e.Main.Humidity,
"pressure": e.Main.Pressure,
"rain": e.Rain.Rain3,
"sunrise": time.Unix(e.Sys.Sunrise, 0).UnixNano(),
"sunset": time.Unix(e.Sys.Sunset, 0).UnixNano(),
"temperature": e.Main.Temp,
"visibility": e.Visibility,
"wind_degrees": e.Wind.Deg,
"wind_speed": e.Wind.Speed,
},
map[string]string{
"city": e.Name,
"city_id": strconv.FormatInt(e.Id, 10),
"country": e.Sys.Country,
"forecast": "*",
},
tm)
}
}
func gatherForecast(acc telegraf.Accumulator, status *Status) {
tags := map[string]string{
"city_id": strconv.FormatInt(status.City.Id, 10),
"forecast": "*",
"city": status.City.Name,
"country": status.City.Country,
}
for i, e := range status.List {
tm := time.Unix(e.Dt, 0)
tags["forecast"] = fmt.Sprintf("%dh", (i+1)*3)
acc.AddFields(
"weather",
map[string]interface{}{
"cloudiness": e.Clouds.All,
"humidity": e.Main.Humidity,
"pressure": e.Main.Pressure,
"rain": e.Rain.Rain3,
"temperature": e.Main.Temp,
"wind_degrees": e.Wind.Deg,
"wind_speed": e.Wind.Speed,
},
tags,
tm)
}
if forecast {
// intentional: overwrite future data points
// under the * tag
tags := map[string]string{
"city_id": strconv.FormatInt(s.City.Id, 10),
"forecast": "*",
}
for _, e := range s.List {
tm := time.Unix(e.Dt, 0)
if e.Id > 0 {
tags["city_id"] = strconv.FormatInt(e.Id, 10)
}
acc.AddFields(
"weather",
map[string]interface{}{
"rain": e.Rain.Rain3,
"wind_degrees": e.Wind.Deg,
"wind_speed": e.Wind.Speed,
"humidity": e.Main.Humidity,
"pressure": e.Main.Pressure,
"temperature": e.Main.Temp,
},
tags,
tm)
}
}
}
func init() {
@@ -300,6 +311,7 @@ func init() {
return &OpenWeatherMap{
ResponseTimeout: tmout,
Units: defaultUnits,
BaseUrl: defaultBaseURL,
}
})
}