Add fluentd input plugin (#2661)

This commit is contained in:
DanKans 2017-07-13 23:58:20 +01:00 committed by Daniel Nelson
parent 4138e8af08
commit 0ca0e213de
4 changed files with 407 additions and 0 deletions

View File

@ -23,6 +23,7 @@ import (
_ "github.com/influxdata/telegraf/plugins/inputs/exec" _ "github.com/influxdata/telegraf/plugins/inputs/exec"
_ "github.com/influxdata/telegraf/plugins/inputs/fail2ban" _ "github.com/influxdata/telegraf/plugins/inputs/fail2ban"
_ "github.com/influxdata/telegraf/plugins/inputs/filestat" _ "github.com/influxdata/telegraf/plugins/inputs/filestat"
_ "github.com/influxdata/telegraf/plugins/inputs/fluentd"
_ "github.com/influxdata/telegraf/plugins/inputs/graylog" _ "github.com/influxdata/telegraf/plugins/inputs/graylog"
_ "github.com/influxdata/telegraf/plugins/inputs/haproxy" _ "github.com/influxdata/telegraf/plugins/inputs/haproxy"
_ "github.com/influxdata/telegraf/plugins/inputs/hddtemp" _ "github.com/influxdata/telegraf/plugins/inputs/hddtemp"

View File

@ -0,0 +1,64 @@
# Fluentd Input Plugin
The fluentd plugin gathers metrics from plugin endpoint provided by [in_monitor plugin](http://docs.fluentd.org/v0.12/articles/monitoring).
This plugin understands data provided by /api/plugin.json resource (/api/config.json is not covered).
You might need to adjust your fluentd configuration, in order to reduce series cardinality in case whene your fluentd restarts frequently. Every time when fluentd starts, `plugin_id` value is given a new random value.
According to [fluentd documentation](http://docs.fluentd.org/v0.12/articles/config-file), you are able to add `@id` parameter for each plugin to avoid this behaviour and define custom `plugin_id`.
example configuratio with `@id` parameter for http plugin:
```
<source>
@type http
@id http
port 8888
</source>
```
### Configuration:
```toml
# Read metrics exposed by fluentd in_monitor plugin
[[inputs.fluentd]]
## This plugin reads information exposed by fluentd (using /api/plugins.json endpoint).
##
## Endpoint:
## - only one URI is allowed
## - https is not supported
endpoint = "http://localhost:24220/api/plugins.json"
## Define which plugins have to be excluded (based on "type" field - e.g. monitor_agent)
exclude = [
"monitor_agent",
"dummy",
]
```
### Measurements & Fields:
Fields may vary depends on type of the plugin
- fluentd
- retry_count (float, unit)
- buffer_queue_length (float, unit)
- buffer_total_queued_size (float, unit)
### Tags:
- All measurements have the following tags:
- plugin_id (unique plugin id)
- plugin_type (type of the plugin e.g. s3)
- plugin_category (plugin category e.g. output)
### Example Output:
```
$ telegraf --config fluentd.conf --input-filter fluentd --test
* Plugin: inputs.fluentd, Collection 1
> fluentd,host=T440s,plugin_id=object:9f748c,plugin_category=input,plugin_type=dummy buffer_total_queued_size=0,buffer_queue_length=0,retry_count=0 1492006105000000000
> fluentd,plugin_category=input,plugin_type=dummy,host=T440s,plugin_id=object:8da98c buffer_queue_length=0,retry_count=0,buffer_total_queued_size=0 1492006105000000000
> fluentd,plugin_id=object:820190,plugin_category=input,plugin_type=monitor_agent,host=T440s retry_count=0,buffer_total_queued_size=0,buffer_queue_length=0 1492006105000000000
> fluentd,plugin_id=object:c5e054,plugin_category=output,plugin_type=stdout,host=T440s buffer_queue_length=0,retry_count=0,buffer_total_queued_size=0 1492006105000000000
> fluentd,plugin_type=s3,host=T440s,plugin_id=object:bd7a90,plugin_category=output buffer_queue_length=0,retry_count=0,buffer_total_queued_size=0 1492006105000000000
```

View File

@ -0,0 +1,173 @@
package fluentd
import (
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"net/url"
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/inputs"
)
const (
measurement = "fluentd"
description = "Read metrics exposed by fluentd in_monitor plugin"
sampleConfig = `
## This plugin reads information exposed by fluentd (using /api/plugins.json endpoint).
##
## Endpoint:
## - only one URI is allowed
## - https is not supported
endpoint = "http://localhost:24220/api/plugins.json"
## Define which plugins have to be excluded (based on "type" field - e.g. monitor_agent)
exclude = [
"monitor_agent",
"dummy",
]
`
)
// Fluentd - plugin main structure
type Fluentd struct {
Endpoint string
Exclude []string
client *http.Client
}
type endpointInfo struct {
Payload []pluginData `json:"plugins"`
}
type pluginData struct {
PluginID string `json:"plugin_id"`
PluginType string `json:"type"`
PluginCategory string `json:"plugin_category"`
RetryCount *float64 `json:"retry_count"`
BufferQueueLength *float64 `json:"buffer_queue_length"`
BufferTotalQueuedSize *float64 `json:"buffer_total_queued_size"`
}
// parse JSON from fluentd Endpoint
// Parameters:
// data: unprocessed json recivied from endpoint
//
// Returns:
// pluginData: slice that contains parsed plugins
// error: error that may have occurred
func parse(data []byte) (datapointArray []pluginData, err error) {
var endpointData endpointInfo
if err = json.Unmarshal(data, &endpointData); err != nil {
err = fmt.Errorf("Processing JSON structure")
return
}
for _, point := range endpointData.Payload {
datapointArray = append(datapointArray, point)
}
return
}
// Description - display description
func (h *Fluentd) Description() string { return description }
// SampleConfig - generate configuretion
func (h *Fluentd) SampleConfig() string { return sampleConfig }
// Gather - Main code responsible for gathering, processing and creating metrics
func (h *Fluentd) Gather(acc telegraf.Accumulator) error {
_, err := url.Parse(h.Endpoint)
if err != nil {
return fmt.Errorf("Invalid URL \"%s\"", h.Endpoint)
}
if h.client == nil {
tr := &http.Transport{
ResponseHeaderTimeout: time.Duration(3 * time.Second),
}
client := &http.Client{
Transport: tr,
Timeout: time.Duration(4 * time.Second),
}
h.client = client
}
resp, err := h.client.Get(h.Endpoint)
if err != nil {
return fmt.Errorf("Unable to perform HTTP client GET on \"%s\": %s", h.Endpoint, err)
}
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return fmt.Errorf("Unable to read the HTTP body \"%s\": %s", string(body), err)
}
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("http status ok not met")
}
dataPoints, err := parse(body)
if err != nil {
return fmt.Errorf("Problem with parsing")
}
// Go through all plugins one by one
for _, p := range dataPoints {
skip := false
// Check if this specific type was excluded in configuration
for _, exclude := range h.Exclude {
if exclude == p.PluginType {
skip = true
}
}
// If not, create new metric and add it to Accumulator
if !skip {
tmpFields := make(map[string]interface{})
tmpTags := map[string]string{
"plugin_id": p.PluginID,
"plugin_category": p.PluginCategory,
"plugin_type": p.PluginType,
}
if p.BufferQueueLength != nil {
tmpFields["buffer_queue_length"] = p.BufferQueueLength
}
if p.RetryCount != nil {
tmpFields["retry_count"] = p.RetryCount
}
if p.BufferTotalQueuedSize != nil {
tmpFields["buffer_total_queued_size"] = p.BufferTotalQueuedSize
}
if !((p.BufferQueueLength == nil) && (p.RetryCount == nil) && (p.BufferTotalQueuedSize == nil)) {
acc.AddFields(measurement, tmpFields, tmpTags)
}
}
}
return nil
}
func init() {
inputs.Add("fluentd", func() telegraf.Input { return &Fluentd{} })
}

View File

@ -0,0 +1,169 @@
package fluentd
import (
"fmt"
"net"
"net/http"
"net/http/httptest"
"net/url"
"testing"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/assert"
)
// sampleJSON from fluentd version '0.14.9'
const sampleJSON = `
{
"plugins": [
{
"plugin_id": "object:f48698",
"plugin_category": "input",
"type": "dummy",
"config": {
"@type": "dummy",
"@log_level": "info",
"tag": "stdout.page.node",
"rate": "",
"dummy": "{\"hello\":\"world_from_first_dummy\"}",
"auto_increment_key": "id1"
},
"output_plugin": false,
"retry_count": null
},
{
"plugin_id": "object:e27138",
"plugin_category": "input",
"type": "dummy",
"config": {
"@type": "dummy",
"@log_level": "info",
"tag": "stdout.superproject.supercontainer",
"rate": "",
"dummy": "{\"hello\":\"world_from_second_dummy\"}",
"auto_increment_key": "id1"
},
"output_plugin": false,
"retry_count": null
},
{
"plugin_id": "object:d74060",
"plugin_category": "input",
"type": "monitor_agent",
"config": {
"@type": "monitor_agent",
"@log_level": "error",
"bind": "0.0.0.0",
"port": "24220"
},
"output_plugin": false,
"retry_count": null
},
{
"plugin_id": "object:11a5e2c",
"plugin_category": "output",
"type": "stdout",
"config": {
"@type": "stdout"
},
"output_plugin": true,
"retry_count": 0
},
{
"plugin_id": "object:11237ec",
"plugin_category": "output",
"type": "s3",
"config": {
"@type": "s3",
"@log_level": "info",
"aws_key_id": "xxxxxx",
"aws_sec_key": "xxxxxx",
"s3_bucket": "bucket",
"s3_endpoint": "http://mock:4567",
"path": "logs/%Y%m%d_%H/${tag[1]}/",
"time_slice_format": "%M",
"s3_object_key_format": "%{path}%{time_slice}_%{hostname}_%{index}_%{hex_random}.%{file_extension}",
"store_as": "gzip"
},
"output_plugin": true,
"buffer_queue_length": 0,
"buffer_total_queued_size": 0,
"retry_count": 0
}
]
}
`
var (
zero float64
err error
pluginOutput []pluginData
expectedOutput = []pluginData{
// {"object:f48698", "dummy", "input", nil, nil, nil},
// {"object:e27138", "dummy", "input", nil, nil, nil},
// {"object:d74060", "monitor_agent", "input", nil, nil, nil},
{"object:11a5e2c", "stdout", "output", (*float64)(&zero), nil, nil},
{"object:11237ec", "s3", "output", (*float64)(&zero), (*float64)(&zero), (*float64)(&zero)},
}
fluentdTest = &Fluentd{
Endpoint: "http://localhost:8081",
}
)
func Test_parse(t *testing.T) {
t.Log("Testing parser function")
_, err := parse([]byte(sampleJSON))
if err != nil {
t.Error(err)
}
}
func Test_Gather(t *testing.T) {
if testing.Short() {
t.Skip("Skipping Gather function test")
}
t.Log("Testing Gather function")
t.Logf("Start HTTP mock (%s) with sampleJSON", fluentdTest.Endpoint)
ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
fmt.Fprintf(w, "%s", string(sampleJSON))
}))
requestURL, err := url.Parse(fluentdTest.Endpoint)
ts.Listener, _ = net.Listen("tcp", fmt.Sprintf("%s:%s", requestURL.Hostname(), requestURL.Port()))
ts.Start()
defer ts.Close()
var acc testutil.Accumulator
err = fluentdTest.Gather(&acc)
if err != nil {
t.Error(err)
}
if !acc.HasMeasurement("fluentd") {
t.Errorf("acc.HasMeasurement: expected fluentd")
}
assert.Equal(t, expectedOutput[0].PluginID, acc.Metrics[0].Tags["plugin_id"])
assert.Equal(t, expectedOutput[0].PluginType, acc.Metrics[0].Tags["plugin_type"])
assert.Equal(t, expectedOutput[0].PluginCategory, acc.Metrics[0].Tags["plugin_category"])
assert.Equal(t, expectedOutput[0].RetryCount, acc.Metrics[0].Fields["retry_count"])
assert.Equal(t, expectedOutput[1].PluginID, acc.Metrics[1].Tags["plugin_id"])
assert.Equal(t, expectedOutput[1].PluginType, acc.Metrics[1].Tags["plugin_type"])
assert.Equal(t, expectedOutput[1].PluginCategory, acc.Metrics[1].Tags["plugin_category"])
assert.Equal(t, expectedOutput[1].RetryCount, acc.Metrics[1].Fields["retry_count"])
assert.Equal(t, expectedOutput[1].BufferQueueLength, acc.Metrics[1].Fields["buffer_queue_length"])
assert.Equal(t, expectedOutput[1].BufferTotalQueuedSize, acc.Metrics[1].Fields["buffer_total_queued_size"])
}