-
Notifications
You must be signed in to change notification settings - Fork 5.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add NATS Monitoring Input Plugin (#3674)
- Loading branch information
1 parent
49e2308
commit 1a16126
Showing
6 changed files
with
239 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| # NATS Monitoring Input Plugin | ||
|
|
||
| The [NATS](http://www.nats.io/about/) monitoring plugin reads from | ||
| specified NATS instance and submits metrics to InfluxDB. | ||
|
|
||
| ## Configuration | ||
|
|
||
| ```toml | ||
| [[inputs.nats]] | ||
| ## The address of the monitoring end-point of the NATS server | ||
| server = "http://localhost:8222" | ||
| ``` |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| package nats | ||
|
|
||
| import ( | ||
| "io/ioutil" | ||
| "net/http" | ||
| "net/url" | ||
| "path" | ||
| "time" | ||
|
|
||
| "encoding/json" | ||
|
|
||
| "github.com/influxdata/telegraf" | ||
| "github.com/influxdata/telegraf/internal" | ||
| "github.com/influxdata/telegraf/plugins/inputs" | ||
|
|
||
| gnatsd "github.com/nats-io/gnatsd/server" | ||
| ) | ||
|
|
||
| type Nats struct { | ||
| Server string | ||
| ResponseTimeout internal.Duration | ||
|
|
||
| client *http.Client | ||
| } | ||
|
|
||
| var sampleConfig = ` | ||
| ## The address of the monitoring endpoint of the NATS server | ||
| server = "http://localhost:1337" | ||
| ## Maximum time to receive response | ||
| # response_timeout = "5s" | ||
| ` | ||
|
|
||
| func (n *Nats) SampleConfig() string { | ||
| return sampleConfig | ||
| } | ||
|
|
||
| func (n *Nats) Description() string { | ||
| return "Provides metrics about the state of a NATS server" | ||
| } | ||
|
|
||
| func (n *Nats) Gather(acc telegraf.Accumulator) error { | ||
| url, err := url.Parse(n.Server) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| url.Path = path.Join(url.Path, "varz") | ||
|
|
||
| if n.client == nil { | ||
| n.client = n.createHTTPClient() | ||
| } | ||
| resp, err := n.client.Get(url.String()) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| defer resp.Body.Close() | ||
|
|
||
| bytes, err := ioutil.ReadAll(resp.Body) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| stats := new(gnatsd.Varz) | ||
| err = json.Unmarshal([]byte(bytes), &stats) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| acc.AddFields("nats", | ||
| map[string]interface{}{ | ||
| "in_msgs": stats.InMsgs, | ||
| "out_msgs": stats.OutMsgs, | ||
| "in_bytes": stats.InBytes, | ||
| "out_bytes": stats.OutBytes, | ||
| "uptime": stats.Now.Sub(stats.Start).Nanoseconds(), | ||
| "cores": stats.Cores, | ||
| "cpu": stats.CPU, | ||
| "mem": stats.Mem, | ||
| "connections": stats.Connections, | ||
| "total_connections": stats.TotalConnections, | ||
| "subscriptions": stats.Subscriptions, | ||
| "slow_consumers": stats.SlowConsumers, | ||
| "routes": stats.Routes, | ||
| "remotes": stats.Remotes, | ||
| }, | ||
| map[string]string{"server": n.Server}, | ||
| time.Now()) | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (n *Nats) createHTTPClient() *http.Client { | ||
| transport := &http.Transport{ | ||
| Proxy: http.ProxyFromEnvironment, | ||
| } | ||
| timeout := n.ResponseTimeout.Duration | ||
| if timeout == time.Duration(0) { | ||
| timeout = 5 * time.Second | ||
| } | ||
| return &http.Client{ | ||
| Transport: transport, | ||
| Timeout: timeout, | ||
| } | ||
| } | ||
|
|
||
| func init() { | ||
| inputs.Add("nats", func() telegraf.Input { | ||
| return &Nats{ | ||
| Server: "http://localhost:8222", | ||
| } | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| package nats | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "net/http" | ||
| "net/http/httptest" | ||
| "testing" | ||
|
|
||
| "github.com/influxdata/telegraf/testutil" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| var sampleVarz = ` | ||
| { | ||
| "server_id": "n2afhLHLl64Gcaj7S7jaNa", | ||
| "version": "1.0.0", | ||
| "go": "go1.8", | ||
| "host": "0.0.0.0", | ||
| "auth_required": false, | ||
| "ssl_required": false, | ||
| "tls_required": false, | ||
| "tls_verify": false, | ||
| "addr": "0.0.0.0", | ||
| "max_connections": 65536, | ||
| "ping_interval": 120000000000, | ||
| "ping_max": 2, | ||
| "http_host": "0.0.0.0", | ||
| "http_port": 1337, | ||
| "https_port": 0, | ||
| "auth_timeout": 1, | ||
| "max_control_line": 1024, | ||
| "cluster": { | ||
| "addr": "0.0.0.0", | ||
| "cluster_port": 0, | ||
| "auth_timeout": 1 | ||
| }, | ||
| "tls_timeout": 0.5, | ||
| "port": 4222, | ||
| "max_payload": 1048576, | ||
| "start": "1861-04-12T10:15:26.841483489-05:00", | ||
| "now": "2011-10-05T15:24:23.722084098-07:00", | ||
| "uptime": "150y5md237h8m57s", | ||
| "mem": 15581184, | ||
| "cores": 48, | ||
| "cpu": 9, | ||
| "connections": 5, | ||
| "total_connections": 109, | ||
| "routes": 1, | ||
| "remotes": 2, | ||
| "in_msgs": 74148556, | ||
| "out_msgs": 68863261, | ||
| "in_bytes": 946267004717, | ||
| "out_bytes": 948110960598, | ||
| "slow_consumers": 2, | ||
| "subscriptions": 4, | ||
| "http_req_stats": { | ||
| "/": 1, | ||
| "/connz": 100847, | ||
| "/routez": 0, | ||
| "/subsz": 1, | ||
| "/varz": 205785 | ||
| }, | ||
| "config_load_time": "2017-07-24T10:15:26.841483489-05:00" | ||
| } | ||
| ` | ||
|
|
||
| func TestMetricsCorrect(t *testing.T) { | ||
| var acc testutil.Accumulator | ||
|
|
||
| srv := newTestNatsServer() | ||
| defer srv.Close() | ||
|
|
||
| n := &Nats{Server: srv.URL} | ||
| err := n.Gather(&acc) | ||
| require.NoError(t, err) | ||
|
|
||
| fields := map[string]interface{}{ | ||
| "in_msgs": int64(74148556), | ||
| "out_msgs": int64(68863261), | ||
| "in_bytes": int64(946267004717), | ||
| "out_bytes": int64(948110960598), | ||
| "uptime": int64(4748742536880600609), | ||
| "cores": 48, | ||
| "cpu": float64(9), | ||
| "mem": int64(15581184), | ||
| "connections": int(5), | ||
| "total_connections": uint64(109), | ||
| "subscriptions": uint32(4), | ||
| "slow_consumers": int64(2), | ||
| "routes": int(1), | ||
| "remotes": int(2), | ||
| } | ||
| tags := map[string]string{ | ||
| "server": srv.URL, | ||
| } | ||
| acc.AssertContainsTaggedFields(t, "nats", fields, tags) | ||
| } | ||
|
|
||
| func newTestNatsServer() *httptest.Server { | ||
| return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | ||
| var rsp string | ||
|
|
||
| switch r.URL.Path { | ||
| case "/varz": | ||
| rsp = sampleVarz | ||
| default: | ||
| panic("Cannot handle request") | ||
| } | ||
|
|
||
| fmt.Fprintln(w, rsp) | ||
| })) | ||
| } |