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

Rename formdata parser to form_urlencoded

This commit is contained in:
Daniel Nelson 2019-06-17 14:44:25 -07:00
parent fd9abd2166
commit 9b338410cb
No known key found for this signature in database
GPG Key ID: CAAD59C9444F6155
8 changed files with 107 additions and 140 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
internal/config/config.go
View File

@ -1726,12 +1726,12 @@ func getParserConfig(name string, tbl *ast.Table) (*parsers.Config, error) {
} }
} }
if node, ok := tbl.Fields["form_data_tag_keys"]; ok { if node, ok := tbl.Fields["form_urlencoded_tag_keys"]; ok {
if kv, ok := node.(*ast.KeyValue); ok { if kv, ok := node.(*ast.KeyValue); ok {
if ary, ok := kv.Value.(*ast.Array); ok { if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value { for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok { if str, ok := elem.(*ast.String); ok {
c.FormDataTagKeys = append(c.FormDataTagKeys, str.Value) c.FormUrlencodedTagKeys = append(c.FormUrlencodedTagKeys, str.Value)
} }
} }
} }
@ -1779,7 +1779,7 @@ func getParserConfig(name string, tbl *ast.Table) (*parsers.Config, error) {
delete(tbl.Fields, "csv_timestamp_column") delete(tbl.Fields, "csv_timestamp_column")
delete(tbl.Fields, "csv_timestamp_format") delete(tbl.Fields, "csv_timestamp_format")
delete(tbl.Fields, "csv_trim_space") delete(tbl.Fields, "csv_trim_space")
delete(tbl.Fields, "form_data_tag_keys") delete(tbl.Fields, "form_urlencoded_tag_keys")
return c, nil return c, nil
} }

10
plugins/inputs/http_listener_v2/README.md
View File

@ -31,6 +31,10 @@ This is a sample configuration for the plugin.
## 0 means to use the default of 524,288,000 bytes (500 mebibytes) ## 0 means to use the default of 524,288,000 bytes (500 mebibytes)
# max_body_size = "500MB" # max_body_size = "500MB"
## Part of the request to consume. Available options are "body" and
## "query".
# data_source = "body"
## Set one or more allowed client CA certificate file names to ## Set one or more allowed client CA certificate file names to
## enable mutually authenticated TLS connections ## enable mutually authenticated TLS connections
# tls_allowed_cacerts = ["/etc/telegraf/clientca.pem"] # tls_allowed_cacerts = ["/etc/telegraf/clientca.pem"]
@ -49,12 +53,6 @@ This is a sample configuration for the plugin.
## more about them here: ## more about them here:
## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
data_format = "influx" data_format = "influx"
## Part of the request to consume.
## Available options are "body" and "query".
## Note that the data source and data format are independent properties.
## To consume standard query params and POST forms - use "formdata" as a data_format.
# data_source = "body"
``` ```
### Metrics: ### Metrics:

43
plugins/inputs/http_listener_v2/http_listener_v2.go
View File

@ -35,21 +35,18 @@ type TimeFunc func() time.Time
// HTTPListenerV2 is an input plugin that collects external metrics sent via HTTP // HTTPListenerV2 is an input plugin that collects external metrics sent via HTTP
type HTTPListenerV2 struct { type HTTPListenerV2 struct {
ServiceAddress string ServiceAddress string `toml:"service_address"`
Path string Path string `toml:"path"`
Methods []string Methods []string `toml:"methods"`
DataSource string DataSource string `toml:"data_source"`
ReadTimeout internal.Duration `toml:"read_timeout"`
ReadTimeout internal.Duration WriteTimeout internal.Duration `toml:"write_timeout"`
WriteTimeout internal.Duration MaxBodySize internal.Size `toml:"max_body_size"`
MaxBodySize internal.Size Port int `toml:"port"`
Port int BasicUsername string `toml:"basic_username"`
BasicPassword string `toml:"basic_password"`
tlsint.ServerConfig tlsint.ServerConfig
BasicUsername string
BasicPassword string
TimeFunc TimeFunc
wg sync.WaitGroup wg sync.WaitGroup
@ -79,6 +76,10 @@ const sampleConfig = `
## 0 means to use the default of 524,288,00 bytes (500 mebibytes) ## 0 means to use the default of 524,288,00 bytes (500 mebibytes)
# max_body_size = "500MB" # max_body_size = "500MB"
## Part of the request to consume. Available options are "body" and
## "query".
# data_source = "body"
## Set one or more allowed client CA certificate file names to ## Set one or more allowed client CA certificate file names to
## enable mutually authenticated TLS connections ## enable mutually authenticated TLS connections
# tls_allowed_cacerts = ["/etc/telegraf/clientca.pem"] # tls_allowed_cacerts = ["/etc/telegraf/clientca.pem"]
@ -97,12 +98,6 @@ const sampleConfig = `
## more about them here: ## more about them here:
## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
data_format = "influx" data_format = "influx"
## Part of the request to consume.
## Available options are "body" and "query".
## Note that the data source and data format are independent properties.
## To consume standard query params and POST forms - use "formdata" as a data_format.
# data_source = "body"
` `
func (h *HTTPListenerV2) SampleConfig() string { func (h *HTTPListenerV2) SampleConfig() string {
@ -167,7 +162,7 @@ func (h *HTTPListenerV2) Start(acc telegraf.Accumulator) error {
server.Serve(h.listener) server.Serve(h.listener)
}() }()
log.Printf("I! Started HTTP listener V2 service on %s\n", h.ServiceAddress) log.Printf("I! [inputs.http_listener_v2] Listening on %s", listener.Addr().String())
return nil return nil
} }
@ -176,8 +171,6 @@ func (h *HTTPListenerV2) Start(acc telegraf.Accumulator) error {
func (h *HTTPListenerV2) Stop() { func (h *HTTPListenerV2) Stop() {
h.listener.Close() h.listener.Close()
h.wg.Wait() h.wg.Wait()
log.Println("I! Stopped HTTP listener V2 service on ", h.ServiceAddress)
} }
func (h *HTTPListenerV2) ServeHTTP(res http.ResponseWriter, req *http.Request) { func (h *HTTPListenerV2) ServeHTTP(res http.ResponseWriter, req *http.Request) {
@ -226,13 +219,13 @@ func (h *HTTPListenerV2) serveWrite(res http.ResponseWriter, req *http.Request)
metrics, err := h.Parse(bytes) metrics, err := h.Parse(bytes)
if err != nil { if err != nil {
log.Println("D! " + err.Error()) log.Printf("D! [inputs.http_listener_v2] Parse error: %v", err)
badRequest(res) badRequest(res)
return return
} }
for _, m := range metrics { for _, m := range metrics {
h.acc.AddFields(m.Name(), m.Fields(), m.Tags(), m.Time()) h.acc.AddMetric(m)
} }
res.WriteHeader(http.StatusNoContent) res.WriteHeader(http.StatusNoContent)
@ -268,7 +261,7 @@ func (h *HTTPListenerV2) collectQuery(res http.ResponseWriter, req *http.Request
query, err := url.QueryUnescape(rawQuery) query, err := url.QueryUnescape(rawQuery)
if err != nil { if err != nil {
log.Println("D! " + err.Error()) log.Printf("D! [inputs.http_listener_v2] Error parsing query: %v", err)
badRequest(res) badRequest(res)
return nil, false return nil, false
} }

57
plugins/parsers/form_urlencoded/README.md Normal file
View File

@ -0,0 +1,57 @@
# Form Urlencoded
The `form-urlencoded` data format parses `application/x-www-form-urlencoded`
data, such as commonly used in the [query string][].
A common use case is to pair it with [http_listener_v2][] input plugin to parse
request body or query params.
### Configuration
```toml
[[inputs.http_listener_v2]]
## Address and port to host HTTP listener on
service_address = ":8080"
## Part of the request to consume. Available options are "body" and
## "query".
data_source = "body"
## Data format to consume.
## Each data format has its own unique set of configuration options, read
## more about them here:
## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
data_format = "form_urlencoded"
## Array of key names which should be collected as tags.
## By default, keys with string value are ignored if not marked as tags.
form_urlencoded_tag_keys = ["tag1"]
```
### Examples
#### Basic parsing
Config:
```toml
[[inputs.http_listener_v2]]
name_override = "mymetric"
service_address = ":8080"
data_source = "query"
data_format = "form_urlencoded"
form_urlencoded_tag_keys = ["tag1"]
```
Request:
```bash
curl -i -XGET 'http://localhost:8080/telegraf?tag1=foo&field1=0.42&field2=42'
```
Output:
```
mymetric,tag1=foo field1=0.42,field2=42
```
[query_string]: https://en.wikipedia.org/wiki/Query_string
[http_listener_v2]: /plugins/inputs/http_listener_v2

7
plugins/parsers/formdata/parser.go → plugins/parsers/form_urlencoded/parser.go
View File

@ -1,4 +1,4 @@
package formdata package form_urlencoded
import ( import (
"bytes" "bytes"
@ -32,7 +32,6 @@ func (p Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
} }
values, err := url.ParseQuery(string(buf)) values, err := url.ParseQuery(string(buf))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -49,7 +48,6 @@ func (p Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
} }
metric, err := metric.New(p.MetricName, tags, fields, time.Now().UTC()) metric, err := metric.New(p.MetricName, tags, fields, time.Now().UTC())
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -60,7 +58,6 @@ func (p Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
// ParseLine delegates a single line of text to the Parse function // ParseLine delegates a single line of text to the Parse function
func (p Parser) ParseLine(line string) (telegraf.Metric, error) { func (p Parser) ParseLine(line string) (telegraf.Metric, error) {
metrics, err := p.Parse([]byte(line)) metrics, err := p.Parse([]byte(line))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -82,7 +79,6 @@ func (p Parser) filterAllowedKeys(original url.Values) url.Values {
for _, key := range p.AllowedKeys { for _, key := range p.AllowedKeys {
value, exists := original[key] value, exists := original[key]
if !exists { if !exists {
continue continue
} }
@ -118,7 +114,6 @@ func (p Parser) parseFields(values url.Values) map[string]interface{} {
} }
field, err := strconv.ParseFloat(value[0], 64) field, err := strconv.ParseFloat(value[0], 64)
if err != nil { if err != nil {
continue continue
} }

32
plugins/parsers/formdata/parser_test.go → plugins/parsers/form_urlencoded/parser_test.go
View File

@ -1,4 +1,4 @@
package formdata package form_urlencoded
import ( import (
"testing" "testing"
@ -16,13 +16,13 @@ const (
func TestParseValidFormData(t *testing.T) { func TestParseValidFormData(t *testing.T) {
parser := Parser{ parser := Parser{
MetricName: "formdata_test", MetricName: "form_urlencoded_test",
} }
metrics, err := parser.Parse([]byte(validFormData)) metrics, err := parser.Parse([]byte(validFormData))
require.NoError(t, err) require.NoError(t, err)
require.Len(t, metrics, 1) require.Len(t, metrics, 1)
require.Equal(t, "formdata_test", metrics[0].Name()) require.Equal(t, "form_urlencoded_test", metrics[0].Name())
require.Equal(t, map[string]string{}, metrics[0].Tags()) require.Equal(t, map[string]string{}, metrics[0].Tags())
require.Equal(t, map[string]interface{}{ require.Equal(t, map[string]interface{}{
"field1": float64(42), "field1": float64(42),
@ -32,12 +32,12 @@ func TestParseValidFormData(t *testing.T) {
func TestParseLineValidFormData(t *testing.T) { func TestParseLineValidFormData(t *testing.T) {
parser := Parser{ parser := Parser{
MetricName: "formdata_test", MetricName: "form_urlencoded_test",
} }
metric, err := parser.ParseLine(validFormData) metric, err := parser.ParseLine(validFormData)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, "formdata_test", metric.Name()) require.Equal(t, "form_urlencoded_test", metric.Name())
require.Equal(t, map[string]string{}, metric.Tags()) require.Equal(t, map[string]string{}, metric.Tags())
require.Equal(t, map[string]interface{}{ require.Equal(t, map[string]interface{}{
"field1": float64(42), "field1": float64(42),
@ -47,14 +47,14 @@ func TestParseLineValidFormData(t *testing.T) {
func TestParseValidFormDataWithTags(t *testing.T) { func TestParseValidFormDataWithTags(t *testing.T) {
parser := Parser{ parser := Parser{
MetricName: "formdata_test", MetricName: "form_urlencoded_test",
TagKeys: []string{"tag1", "tag2"}, TagKeys: []string{"tag1", "tag2"},
} }
metrics, err := parser.Parse([]byte(validFormData)) metrics, err := parser.Parse([]byte(validFormData))
require.NoError(t, err) require.NoError(t, err)
require.Len(t, metrics, 1) require.Len(t, metrics, 1)
require.Equal(t, "formdata_test", metrics[0].Name()) require.Equal(t, "form_urlencoded_test", metrics[0].Name())
require.Equal(t, map[string]string{ require.Equal(t, map[string]string{
"tag1": "foo", "tag1": "foo",
"tag2": "bar", "tag2": "bar",
@ -67,7 +67,7 @@ func TestParseValidFormDataWithTags(t *testing.T) {
func TestParseValidFormDataDefaultTags(t *testing.T) { func TestParseValidFormDataDefaultTags(t *testing.T) {
parser := Parser{ parser := Parser{
MetricName: "formdata_test", MetricName: "form_urlencoded_test",
TagKeys: []string{"tag1", "tag2"}, TagKeys: []string{"tag1", "tag2"},
DefaultTags: map[string]string{"tag4": "default"}, DefaultTags: map[string]string{"tag4": "default"},
} }
@ -75,7 +75,7 @@ func TestParseValidFormDataDefaultTags(t *testing.T) {
metrics, err := parser.Parse([]byte(validFormData)) metrics, err := parser.Parse([]byte(validFormData))
require.NoError(t, err) require.NoError(t, err)
require.Len(t, metrics, 1) require.Len(t, metrics, 1)
require.Equal(t, "formdata_test", metrics[0].Name()) require.Equal(t, "form_urlencoded_test", metrics[0].Name())
require.Equal(t, map[string]string{ require.Equal(t, map[string]string{
"tag1": "foo", "tag1": "foo",
"tag2": "bar", "tag2": "bar",
@ -89,7 +89,7 @@ func TestParseValidFormDataDefaultTags(t *testing.T) {
func TestParseValidFormDataDefaultTagsOverride(t *testing.T) { func TestParseValidFormDataDefaultTagsOverride(t *testing.T) {
parser := Parser{ parser := Parser{
MetricName: "formdata_test", MetricName: "form_urlencoded_test",
TagKeys: []string{"tag1", "tag2"}, TagKeys: []string{"tag1", "tag2"},
DefaultTags: map[string]string{"tag1": "default"}, DefaultTags: map[string]string{"tag1": "default"},
} }
@ -97,7 +97,7 @@ func TestParseValidFormDataDefaultTagsOverride(t *testing.T) {
metrics, err := parser.Parse([]byte(validFormData)) metrics, err := parser.Parse([]byte(validFormData))
require.NoError(t, err) require.NoError(t, err)
require.Len(t, metrics, 1) require.Len(t, metrics, 1)
require.Equal(t, "formdata_test", metrics[0].Name()) require.Equal(t, "form_urlencoded_test", metrics[0].Name())
require.Equal(t, map[string]string{ require.Equal(t, map[string]string{
"tag1": "default", "tag1": "default",
"tag2": "bar", "tag2": "bar",
@ -110,14 +110,14 @@ func TestParseValidFormDataDefaultTagsOverride(t *testing.T) {
func TestParseEncodedFormData(t *testing.T) { func TestParseEncodedFormData(t *testing.T) {
parser := Parser{ parser := Parser{
MetricName: "formdata_test", MetricName: "form_urlencoded_test",
TagKeys: []string{"tag1"}, TagKeys: []string{"tag1"},
} }
metrics, err := parser.Parse([]byte(encodedFormData)) metrics, err := parser.Parse([]byte(encodedFormData))
require.NoError(t, err) require.NoError(t, err)
require.Len(t, metrics, 1) require.Len(t, metrics, 1)
require.Equal(t, "formdata_test", metrics[0].Name()) require.Equal(t, "form_urlencoded_test", metrics[0].Name())
require.Equal(t, map[string]string{ require.Equal(t, map[string]string{
"tag1": "$$$", "tag1": "$$$",
}, metrics[0].Tags()) }, metrics[0].Tags())
@ -128,7 +128,7 @@ func TestParseEncodedFormData(t *testing.T) {
func TestParseInvalidFormDataError(t *testing.T) { func TestParseInvalidFormDataError(t *testing.T) {
parser := Parser{ parser := Parser{
MetricName: "formdata_test", MetricName: "form_urlencoded_test",
} }
metrics, err := parser.Parse([]byte(notEscapedProperlyFormData)) metrics, err := parser.Parse([]byte(notEscapedProperlyFormData))
@ -138,7 +138,7 @@ func TestParseInvalidFormDataError(t *testing.T) {
func TestParseInvalidFormDataEmptyKey(t *testing.T) { func TestParseInvalidFormDataEmptyKey(t *testing.T) {
parser := Parser{ parser := Parser{
MetricName: "formdata_test", MetricName: "form_urlencoded_test",
} }
// Empty key for field // Empty key for field
@ -163,7 +163,7 @@ func TestParseInvalidFormDataEmptyKey(t *testing.T) {
func TestParseInvalidFormDataEmptyString(t *testing.T) { func TestParseInvalidFormDataEmptyString(t *testing.T) {
parser := Parser{ parser := Parser{
MetricName: "formdata_test", MetricName: "form_urlencoded_test",
} }
metrics, err := parser.Parse([]byte(emptyFormData)) metrics, err := parser.Parse([]byte(emptyFormData))

76
plugins/parsers/formdata/README.md
View File

@ -1,76 +0,0 @@
# FormData
The FormData data format parses a [query string/x-www-form-urlencoded][query_string] data into metric fields.
Common use case is to pair it with http listener input plugin to parse request body or query params.
### Configuration
```toml
[[inputs.http_listener_v2]]
## Address and port to host HTTP listener on
service_address = ":8080"
## Part of the request to consume.
## Available options are "body" and "query".
## To consume standard query params or application/x-www-form-urlencoded body,
## set the data_format option to "formdata".
data_source = "body"
## Data format to consume.
## Each data format has its own unique set of configuration options, read
## more about them here:
## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
data_format = "formdata"
## Array of key names which should be collected as tags.
## By default, keys with string value are ignored if not marked as tags.
form_data_tag_keys = ["tag1"]
```
### Examples
#### Basic parsing
Config:
```toml
[[inputs.http_listener_v2]]
service_address = ":8080"
data_source = "query"
data_format = "formdata"
name_override = "mymetric"
```
Request:
```bash
curl -i -XGET 'http://localhost:8080/telegraf?field=0.42'
```
Output:
```
mymetric field=0.42
```
#### Tags and key filter
Config:
```toml
[[inputs.http_listener_v2]]
service_address = ":8080"
data_source = "query"
data_format = "formdata"
name_override = "mymetric"
fielddrop = ["tag2", "field2"]
form_data_tag_keys = ["tag1"]
```
Request:
```bash
curl -i -XGET 'http://localhost:8080/telegraf?tag1=foo&tag2=bar&field1=42&field2=69'
```
Output:
```
mymetric,tag1=foo field1=42
```
[query_string]: https://en.wikipedia.org/wiki/Query_string

14
plugins/parsers/registry.go
View File

@ -8,7 +8,7 @@ import (
"github.com/influxdata/telegraf/plugins/parsers/collectd" "github.com/influxdata/telegraf/plugins/parsers/collectd"
"github.com/influxdata/telegraf/plugins/parsers/csv" "github.com/influxdata/telegraf/plugins/parsers/csv"
"github.com/influxdata/telegraf/plugins/parsers/dropwizard" "github.com/influxdata/telegraf/plugins/parsers/dropwizard"
"github.com/influxdata/telegraf/plugins/parsers/formdata" "github.com/influxdata/telegraf/plugins/parsers/form_urlencoded"
"github.com/influxdata/telegraf/plugins/parsers/graphite" "github.com/influxdata/telegraf/plugins/parsers/graphite"
"github.com/influxdata/telegraf/plugins/parsers/grok" "github.com/influxdata/telegraf/plugins/parsers/grok"
"github.com/influxdata/telegraf/plugins/parsers/influx" "github.com/influxdata/telegraf/plugins/parsers/influx"
@ -144,7 +144,7 @@ type Config struct {
CSVTrimSpace bool `toml:"csv_trim_space"` CSVTrimSpace bool `toml:"csv_trim_space"`
// FormData configuration // FormData configuration
FormDataTagKeys []string `toml:"form_data_tag_keys"` FormUrlencodedTagKeys []string `toml:"form_urlencoded_tag_keys"`
} }
// NewParser returns a Parser interface based on the given config. // NewParser returns a Parser interface based on the given config.
@ -213,11 +213,11 @@ func NewParser(config *Config) (Parser, error) {
config.DefaultTags) config.DefaultTags)
case "logfmt": case "logfmt":
parser, err = NewLogFmtParser(config.MetricName, config.DefaultTags) parser, err = NewLogFmtParser(config.MetricName, config.DefaultTags)
case "formdata": case "form_urlencoded":
parser, err = NewFormDataParser( parser, err = NewFormUrlencodedParser(
config.MetricName, config.MetricName,
config.DefaultTags, config.DefaultTags,
config.FormDataTagKeys, config.FormUrlencodedTagKeys,
) )
default: default:
err = fmt.Errorf("Invalid data format: %s", config.DataFormat) err = fmt.Errorf("Invalid data format: %s", config.DataFormat)
@ -411,12 +411,12 @@ func NewWavefrontParser(defaultTags map[string]string) (Parser, error) {
return wavefront.NewWavefrontParser(defaultTags), nil return wavefront.NewWavefrontParser(defaultTags), nil
} }
func NewFormDataParser( func NewFormUrlencodedParser(
metricName string, metricName string,
defaultTags map[string]string, defaultTags map[string]string,
tagKeys []string, tagKeys []string,
) (Parser, error) { ) (Parser, error) {
return &formdata.Parser{ return &form_urlencoded.Parser{
MetricName: metricName, MetricName: metricName,
DefaultTags: defaultTags, DefaultTags: defaultTags,
TagKeys: tagKeys, TagKeys: tagKeys,