feat(http): add accept parsing for telegraf
Chris Goller
parent
d00dd1b2da
commit
4f38c8cad9
1
go.mod
1
go.mod
|
|
@ -51,6 +51,7 @@ require (
|
|||
github.com/go-test/deep v1.0.1 // indirect
|
||||
github.com/gocql/gocql v0.0.0-20181117210152-33c0e89ca93a // indirect
|
||||
github.com/gogo/protobuf v1.1.1
|
||||
github.com/golang/gddo v0.0.0-20181116215533-9bd4a3295021
|
||||
github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db
|
||||
github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c
|
||||
github.com/google/go-cmp v0.2.0
|
||||
|
|
|
|||
2
go.sum
2
go.sum
|
|
@ -140,6 +140,8 @@ github.com/gocql/gocql v0.0.0-20181117210152-33c0e89ca93a h1:B5gyGsJJmFKS7exambl
|
|||
github.com/gocql/gocql v0.0.0-20181117210152-33c0e89ca93a/go.mod h1:4Fw1eo5iaEhDUs8XyuhSVCVy52Jq3L+/3GJgYkwc+/0=
|
||||
github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo=
|
||||
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
|
||||
github.com/golang/gddo v0.0.0-20181116215533-9bd4a3295021 h1:HYV500jCgk+IC68L5sWrLFIWMpaUFfXXpJSAb7XOoBk=
|
||||
github.com/golang/gddo v0.0.0-20181116215533-9bd4a3295021/go.mod h1:xEhNfoBDX1hzLm2Nf80qUvZ2sVwoMZ8d6IE2SrsQfh4=
|
||||
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=
|
||||
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
|
||||
github.com/golang/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:tluoj9z5200jBnyusfRPU2LqT6J+DAorxEvtC7LHB+E=
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import (
|
|||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/golang/gddo/httputil"
|
||||
"github.com/influxdata/platform"
|
||||
pctx "github.com/influxdata/platform/context"
|
||||
"github.com/influxdata/platform/kit/errors"
|
||||
|
|
@ -152,7 +153,9 @@ func (h *TelegrafHandler) handleGetTelegraf(w http.ResponseWriter, r *http.Reque
|
|||
return
|
||||
}
|
||||
|
||||
mimeType := r.Header.Get("Accept")
|
||||
offers := []string{"application/toml", "application/json", "application/octet-stream"}
|
||||
defaultOffer := "application/toml"
|
||||
mimeType := httputil.NegotiateContentType(r, offers, defaultOffer)
|
||||
switch mimeType {
|
||||
case "application/octet-stream":
|
||||
w.Header().Set("Content-Type", "application/octet-stream")
|
||||
|
|
@ -164,7 +167,7 @@ func (h *TelegrafHandler) handleGetTelegraf(w http.ResponseWriter, r *http.Reque
|
|||
logEncodingError(h.Logger, r, err)
|
||||
return
|
||||
}
|
||||
default:
|
||||
case "application/toml":
|
||||
w.Header().Set("Content-Type", "application/toml; charset=utf-8")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte(tc.TOML()))
|
||||
|
|
|
|||
|
|
@ -0,0 +1,815 @@
|
|||
package http
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/influxdata/platform"
|
||||
"github.com/influxdata/platform/mock"
|
||||
"github.com/influxdata/platform/telegraf/plugins/inputs"
|
||||
"github.com/influxdata/platform/telegraf/plugins/outputs"
|
||||
"go.uber.org/zap/zaptest"
|
||||
)
|
||||
|
||||
func TestTelegrafHandler_handleGetTelegrafs(t *testing.T) {
|
||||
type wants struct {
|
||||
statusCode int
|
||||
contentType string
|
||||
body string
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
svc *mock.TelegrafConfigStore
|
||||
r *http.Request
|
||||
wants wants
|
||||
}{
|
||||
{
|
||||
name: "return CPU plugin for telegraf",
|
||||
r: httptest.NewRequest("GET", "http://any.url/api/v2/telegrafs", nil),
|
||||
svc: &mock.TelegrafConfigStore{
|
||||
FindTelegrafConfigsF: func(ctx context.Context, filter platform.UserResourceMappingFilter, opt ...platform.FindOptions) ([]*platform.TelegrafConfig, int, error) {
|
||||
return []*platform.TelegrafConfig{
|
||||
&platform.TelegrafConfig{
|
||||
ID: platform.ID(1),
|
||||
Name: "my config",
|
||||
Agent: platform.TelegrafAgentConfig{
|
||||
Interval: 10000,
|
||||
},
|
||||
Plugins: []platform.TelegrafPlugin{
|
||||
{
|
||||
Comment: "my cpu stats",
|
||||
Config: &inputs.CPUStats{},
|
||||
},
|
||||
{
|
||||
Comment: "my influx output",
|
||||
Config: &outputs.InfluxDBV2{
|
||||
URLs: []string{"http://127.0.0.1:9999"},
|
||||
Token: "no_more_secrets",
|
||||
Organization: "my_org",
|
||||
Bucket: "my_bucket",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}, 1, nil
|
||||
},
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusOK,
|
||||
contentType: "application/json; charset=utf-8",
|
||||
// TODO(goller): once links are in for telegraf, this will need to change.
|
||||
body: `{
|
||||
"configurations": [
|
||||
{
|
||||
"id": "0000000000000001",
|
||||
"name": "my config",
|
||||
"agent": {
|
||||
"collectionInterval": 10000
|
||||
},
|
||||
"plugins": [
|
||||
{
|
||||
"name": "cpu",
|
||||
"type": "input",
|
||||
"comment": "my cpu stats",
|
||||
"config": {}
|
||||
},
|
||||
{
|
||||
"name": "influxdb_v2",
|
||||
"type": "output",
|
||||
"comment": "my influx output",
|
||||
"config": {
|
||||
"urls": [
|
||||
"http://127.0.0.1:9999"
|
||||
],
|
||||
"token": "no_more_secrets",
|
||||
"organization": "my_org",
|
||||
"bucket": "my_bucket"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}`,
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
logger := zaptest.NewLogger(t)
|
||||
mapping := mock.NewUserResourceMappingService()
|
||||
labels := mock.NewLabelService()
|
||||
users := mock.NewUserService()
|
||||
w := httptest.NewRecorder()
|
||||
h := NewTelegrafHandler(logger, mapping, labels, tt.svc, users)
|
||||
h.ServeHTTP(w, tt.r)
|
||||
|
||||
res := w.Result()
|
||||
content := res.Header.Get("Content-Type")
|
||||
body, _ := ioutil.ReadAll(res.Body)
|
||||
|
||||
if res.StatusCode != tt.wants.statusCode {
|
||||
t.Errorf("%q. handleGetTelegrafs() = %v, want %v", tt.name, res.StatusCode, tt.wants.statusCode)
|
||||
t.Logf("headers: %v", res.Header)
|
||||
}
|
||||
if tt.wants.contentType != "" && content != tt.wants.contentType {
|
||||
t.Errorf("%q. handleGetTelegrafs() = %v, want %v", tt.name, content, tt.wants.contentType)
|
||||
}
|
||||
if eq, _ := jsonEqual(string(body), tt.wants.body); tt.wants.body != "" && !eq {
|
||||
t.Errorf("%q. handleGetTelegrafs() = \n***%v***\n,\nwant\n***%v***", tt.name, string(body), tt.wants.body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTelegrafHandler_handleGetTelegraf(t *testing.T) {
|
||||
type wants struct {
|
||||
statusCode int
|
||||
contentType string
|
||||
body string
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
svc *mock.TelegrafConfigStore
|
||||
r *http.Request
|
||||
acceptHeader string
|
||||
wants wants
|
||||
}{
|
||||
{
|
||||
name: "return JSON telegraf config",
|
||||
r: httptest.NewRequest("GET", "http://any.url/api/v2/telegrafs/0000000000000001", nil),
|
||||
acceptHeader: "application/json",
|
||||
svc: &mock.TelegrafConfigStore{
|
||||
FindTelegrafConfigByIDF: func(ctx context.Context, id platform.ID) (*platform.TelegrafConfig, error) {
|
||||
return &platform.TelegrafConfig{
|
||||
ID: platform.ID(1),
|
||||
Name: "my config",
|
||||
Agent: platform.TelegrafAgentConfig{
|
||||
Interval: 10000,
|
||||
},
|
||||
Plugins: []platform.TelegrafPlugin{
|
||||
{
|
||||
Comment: "my cpu stats",
|
||||
Config: &inputs.CPUStats{},
|
||||
},
|
||||
{
|
||||
Comment: "my influx output",
|
||||
Config: &outputs.InfluxDBV2{
|
||||
URLs: []string{"http://127.0.0.1:9999"},
|
||||
Token: "no_more_secrets",
|
||||
Organization: "my_org",
|
||||
Bucket: "my_bucket",
|
||||
},
|
||||
},
|
||||
},
|
||||
}, nil
|
||||
},
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusOK,
|
||||
contentType: "application/json; charset=utf-8",
|
||||
// TODO(goller): once links are in for telegraf, this will need to change.
|
||||
body: `{
|
||||
"id": "0000000000000001",
|
||||
"name": "my config",
|
||||
"agent": {
|
||||
"collectionInterval": 10000
|
||||
},
|
||||
"plugins": [
|
||||
{
|
||||
"name": "cpu",
|
||||
"type": "input",
|
||||
"comment": "my cpu stats",
|
||||
"config": {}
|
||||
},
|
||||
{
|
||||
"name": "influxdb_v2",
|
||||
"type": "output",
|
||||
"comment": "my influx output",
|
||||
"config": {
|
||||
"urls": [
|
||||
"http://127.0.0.1:9999"
|
||||
],
|
||||
"token": "no_more_secrets",
|
||||
"organization": "my_org",
|
||||
"bucket": "my_bucket"
|
||||
}
|
||||
}
|
||||
]
|
||||
}`,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "return JSON telegraf config using fuzzy accept header matching",
|
||||
r: httptest.NewRequest("GET", "http://any.url/api/v2/telegrafs/0000000000000001", nil),
|
||||
acceptHeader: "application/json, text/plain, */*",
|
||||
svc: &mock.TelegrafConfigStore{
|
||||
FindTelegrafConfigByIDF: func(ctx context.Context, id platform.ID) (*platform.TelegrafConfig, error) {
|
||||
return &platform.TelegrafConfig{
|
||||
ID: platform.ID(1),
|
||||
Name: "my config",
|
||||
Agent: platform.TelegrafAgentConfig{
|
||||
Interval: 10000,
|
||||
},
|
||||
Plugins: []platform.TelegrafPlugin{
|
||||
{
|
||||
Comment: "my cpu stats",
|
||||
Config: &inputs.CPUStats{},
|
||||
},
|
||||
{
|
||||
Comment: "my influx output",
|
||||
Config: &outputs.InfluxDBV2{
|
||||
URLs: []string{"http://127.0.0.1:9999"},
|
||||
Token: "no_more_secrets",
|
||||
Organization: "my_org",
|
||||
Bucket: "my_bucket",
|
||||
},
|
||||
},
|
||||
},
|
||||
}, nil
|
||||
},
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusOK,
|
||||
contentType: "application/json; charset=utf-8",
|
||||
// TODO(goller): once links are in for telegraf, this will need to change.
|
||||
body: `{
|
||||
"id": "0000000000000001",
|
||||
"name": "my config",
|
||||
"agent": {
|
||||
"collectionInterval": 10000
|
||||
},
|
||||
"plugins": [
|
||||
{
|
||||
"name": "cpu",
|
||||
"type": "input",
|
||||
"comment": "my cpu stats",
|
||||
"config": {}
|
||||
},
|
||||
{
|
||||
"name": "influxdb_v2",
|
||||
"type": "output",
|
||||
"comment": "my influx output",
|
||||
"config": {
|
||||
"urls": [
|
||||
"http://127.0.0.1:9999"
|
||||
],
|
||||
"token": "no_more_secrets",
|
||||
"organization": "my_org",
|
||||
"bucket": "my_bucket"
|
||||
}
|
||||
}
|
||||
]
|
||||
}`,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "return TOML telegraf config with accept header application/toml",
|
||||
r: httptest.NewRequest("GET", "http://any.url/api/v2/telegrafs/0000000000000001", nil),
|
||||
acceptHeader: "application/toml",
|
||||
svc: &mock.TelegrafConfigStore{
|
||||
FindTelegrafConfigByIDF: func(ctx context.Context, id platform.ID) (*platform.TelegrafConfig, error) {
|
||||
return &platform.TelegrafConfig{
|
||||
ID: platform.ID(1),
|
||||
Name: "my config",
|
||||
Agent: platform.TelegrafAgentConfig{
|
||||
Interval: 10000,
|
||||
},
|
||||
Plugins: []platform.TelegrafPlugin{
|
||||
{
|
||||
Comment: "my cpu stats",
|
||||
Config: &inputs.CPUStats{},
|
||||
},
|
||||
{
|
||||
Comment: "my influx output",
|
||||
Config: &outputs.InfluxDBV2{
|
||||
URLs: []string{"http://127.0.0.1:9999"},
|
||||
Token: "no_more_secrets",
|
||||
Organization: "my_org",
|
||||
Bucket: "my_bucket",
|
||||
},
|
||||
},
|
||||
},
|
||||
}, nil
|
||||
},
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusOK,
|
||||
contentType: "application/toml; charset=utf-8",
|
||||
// TODO(goller): once links are in for telegraf, this will need to change.
|
||||
body: `# Configuration for telegraf agent
|
||||
[agent]
|
||||
## Default data collection interval for all inputs
|
||||
interval = "10s"
|
||||
## Rounds collection interval to 'interval'
|
||||
## ie, if interval="10s" then always collect on :00, :10, :20, etc.
|
||||
round_interval = true
|
||||
|
||||
## Telegraf will send metrics to outputs in batches of at most
|
||||
## metric_batch_size metrics.
|
||||
## This controls the size of writes that Telegraf sends to output plugins.
|
||||
metric_batch_size = 1000
|
||||
|
||||
## For failed writes, telegraf will cache metric_buffer_limit metrics for each
|
||||
## output, and will flush this buffer on a successful write. Oldest metrics
|
||||
## are dropped first when this buffer fills.
|
||||
## This buffer only fills when writes fail to output plugin(s).
|
||||
metric_buffer_limit = 10000
|
||||
|
||||
## Collection jitter is used to jitter the collection by a random amount.
|
||||
## Each plugin will sleep for a random time within jitter before collecting.
|
||||
## This can be used to avoid many plugins querying things like sysfs at the
|
||||
## same time, which can have a measurable effect on the system.
|
||||
collection_jitter = "0s"
|
||||
|
||||
## Default flushing interval for all outputs. Maximum flush_interval will be
|
||||
## flush_interval + flush_jitter
|
||||
flush_interval = "10s"
|
||||
## Jitter the flush interval by a random amount. This is primarily to avoid
|
||||
## large write spikes for users running a large number of telegraf instances.
|
||||
## ie, a jitter of 5s and interval 10s means flushes will happen every 10-15s
|
||||
flush_jitter = "0s"
|
||||
|
||||
## By default or when set to "0s", precision will be set to the same
|
||||
## timestamp order as the collection interval, with the maximum being 1s.
|
||||
## ie, when interval = "10s", precision will be "1s"
|
||||
## when interval = "250ms", precision will be "1ms"
|
||||
## Precision will NOT be used for service inputs. It is up to each individual
|
||||
## service input to set the timestamp at the appropriate precision.
|
||||
## Valid time units are "ns", "us" (or "µs"), "ms", "s".
|
||||
precision = ""
|
||||
|
||||
## Logging configuration:
|
||||
## Run telegraf with debug log messages.
|
||||
debug = false
|
||||
## Run telegraf in quiet mode (error log messages only).
|
||||
quiet = false
|
||||
## Specify the log file name. The empty string means to log to stderr.
|
||||
logfile = ""
|
||||
|
||||
## Override default hostname, if empty use os.Hostname()
|
||||
hostname = ""
|
||||
## If set to true, do no set the "host" tag in the telegraf agent.
|
||||
omit_hostname = false
|
||||
[[inputs.cpu]]
|
||||
## Whether to report per-cpu stats or not
|
||||
percpu = true
|
||||
## Whether to report total system cpu stats or not
|
||||
totalcpu = true
|
||||
## If true, collect raw CPU time metrics.
|
||||
collect_cpu_time = false
|
||||
## If true, compute and report the sum of all non-idle CPU states.
|
||||
report_active = false
|
||||
[[outputs.influxdb_v2]]
|
||||
## The URLs of the InfluxDB cluster nodes.
|
||||
##
|
||||
## Multiple URLs can be specified for a single cluster, only ONE of the
|
||||
## urls will be written to each interval.
|
||||
## urls exp: http://127.0.0.1:9999
|
||||
urls = ["http://127.0.0.1:9999"]
|
||||
|
||||
## Token for authentication.
|
||||
token = "no_more_secrets"
|
||||
|
||||
## Organization is the name of the organization you wish to write to; must exist.
|
||||
organization = "my_org"
|
||||
|
||||
## Destination bucket to write into.
|
||||
bucket = "my_bucket"
|
||||
`,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "return TOML telegraf config with no accept header",
|
||||
r: httptest.NewRequest("GET", "http://any.url/api/v2/telegrafs/0000000000000001", nil),
|
||||
svc: &mock.TelegrafConfigStore{
|
||||
FindTelegrafConfigByIDF: func(ctx context.Context, id platform.ID) (*platform.TelegrafConfig, error) {
|
||||
return &platform.TelegrafConfig{
|
||||
ID: platform.ID(1),
|
||||
Name: "my config",
|
||||
Agent: platform.TelegrafAgentConfig{
|
||||
Interval: 10000,
|
||||
},
|
||||
Plugins: []platform.TelegrafPlugin{
|
||||
{
|
||||
Comment: "my cpu stats",
|
||||
Config: &inputs.CPUStats{},
|
||||
},
|
||||
{
|
||||
Comment: "my influx output",
|
||||
Config: &outputs.InfluxDBV2{
|
||||
URLs: []string{"http://127.0.0.1:9999"},
|
||||
Token: "no_more_secrets",
|
||||
Organization: "my_org",
|
||||
Bucket: "my_bucket",
|
||||
},
|
||||
},
|
||||
},
|
||||
}, nil
|
||||
},
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusOK,
|
||||
contentType: "application/toml; charset=utf-8",
|
||||
// TODO(goller): once links are in for telegraf, this will need to change.
|
||||
body: `# Configuration for telegraf agent
|
||||
[agent]
|
||||
## Default data collection interval for all inputs
|
||||
interval = "10s"
|
||||
## Rounds collection interval to 'interval'
|
||||
## ie, if interval="10s" then always collect on :00, :10, :20, etc.
|
||||
round_interval = true
|
||||
|
||||
## Telegraf will send metrics to outputs in batches of at most
|
||||
## metric_batch_size metrics.
|
||||
## This controls the size of writes that Telegraf sends to output plugins.
|
||||
metric_batch_size = 1000
|
||||
|
||||
## For failed writes, telegraf will cache metric_buffer_limit metrics for each
|
||||
## output, and will flush this buffer on a successful write. Oldest metrics
|
||||
## are dropped first when this buffer fills.
|
||||
## This buffer only fills when writes fail to output plugin(s).
|
||||
metric_buffer_limit = 10000
|
||||
|
||||
## Collection jitter is used to jitter the collection by a random amount.
|
||||
## Each plugin will sleep for a random time within jitter before collecting.
|
||||
## This can be used to avoid many plugins querying things like sysfs at the
|
||||
## same time, which can have a measurable effect on the system.
|
||||
collection_jitter = "0s"
|
||||
|
||||
## Default flushing interval for all outputs. Maximum flush_interval will be
|
||||
## flush_interval + flush_jitter
|
||||
flush_interval = "10s"
|
||||
## Jitter the flush interval by a random amount. This is primarily to avoid
|
||||
## large write spikes for users running a large number of telegraf instances.
|
||||
## ie, a jitter of 5s and interval 10s means flushes will happen every 10-15s
|
||||
flush_jitter = "0s"
|
||||
|
||||
## By default or when set to "0s", precision will be set to the same
|
||||
## timestamp order as the collection interval, with the maximum being 1s.
|
||||
## ie, when interval = "10s", precision will be "1s"
|
||||
## when interval = "250ms", precision will be "1ms"
|
||||
## Precision will NOT be used for service inputs. It is up to each individual
|
||||
## service input to set the timestamp at the appropriate precision.
|
||||
## Valid time units are "ns", "us" (or "µs"), "ms", "s".
|
||||
precision = ""
|
||||
|
||||
## Logging configuration:
|
||||
## Run telegraf with debug log messages.
|
||||
debug = false
|
||||
## Run telegraf in quiet mode (error log messages only).
|
||||
quiet = false
|
||||
## Specify the log file name. The empty string means to log to stderr.
|
||||
logfile = ""
|
||||
|
||||
## Override default hostname, if empty use os.Hostname()
|
||||
hostname = ""
|
||||
## If set to true, do no set the "host" tag in the telegraf agent.
|
||||
omit_hostname = false
|
||||
[[inputs.cpu]]
|
||||
## Whether to report per-cpu stats or not
|
||||
percpu = true
|
||||
## Whether to report total system cpu stats or not
|
||||
totalcpu = true
|
||||
## If true, collect raw CPU time metrics.
|
||||
collect_cpu_time = false
|
||||
## If true, compute and report the sum of all non-idle CPU states.
|
||||
report_active = false
|
||||
[[outputs.influxdb_v2]]
|
||||
## The URLs of the InfluxDB cluster nodes.
|
||||
##
|
||||
## Multiple URLs can be specified for a single cluster, only ONE of the
|
||||
## urls will be written to each interval.
|
||||
## urls exp: http://127.0.0.1:9999
|
||||
urls = ["http://127.0.0.1:9999"]
|
||||
|
||||
## Token for authentication.
|
||||
token = "no_more_secrets"
|
||||
|
||||
## Organization is the name of the organization you wish to write to; must exist.
|
||||
organization = "my_org"
|
||||
|
||||
## Destination bucket to write into.
|
||||
bucket = "my_bucket"
|
||||
`,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "return TOML telegraf config with application/octet-stream",
|
||||
r: httptest.NewRequest("GET", "http://any.url/api/v2/telegrafs/0000000000000001", nil),
|
||||
acceptHeader: "application/octet-stream",
|
||||
svc: &mock.TelegrafConfigStore{
|
||||
FindTelegrafConfigByIDF: func(ctx context.Context, id platform.ID) (*platform.TelegrafConfig, error) {
|
||||
return &platform.TelegrafConfig{
|
||||
ID: platform.ID(1),
|
||||
Name: "my config",
|
||||
Agent: platform.TelegrafAgentConfig{
|
||||
Interval: 10000,
|
||||
},
|
||||
Plugins: []platform.TelegrafPlugin{
|
||||
{
|
||||
Comment: "my cpu stats",
|
||||
Config: &inputs.CPUStats{},
|
||||
},
|
||||
{
|
||||
Comment: "my influx output",
|
||||
Config: &outputs.InfluxDBV2{
|
||||
URLs: []string{"http://127.0.0.1:9999"},
|
||||
Token: "no_more_secrets",
|
||||
Organization: "my_org",
|
||||
Bucket: "my_bucket",
|
||||
},
|
||||
},
|
||||
},
|
||||
}, nil
|
||||
},
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusOK,
|
||||
contentType: "application/octet-stream",
|
||||
// TODO(goller): once links are in for telegraf, this will need to change.
|
||||
body: `# Configuration for telegraf agent
|
||||
[agent]
|
||||
## Default data collection interval for all inputs
|
||||
interval = "10s"
|
||||
## Rounds collection interval to 'interval'
|
||||
## ie, if interval="10s" then always collect on :00, :10, :20, etc.
|
||||
round_interval = true
|
||||
|
||||
## Telegraf will send metrics to outputs in batches of at most
|
||||
## metric_batch_size metrics.
|
||||
## This controls the size of writes that Telegraf sends to output plugins.
|
||||
metric_batch_size = 1000
|
||||
|
||||
## For failed writes, telegraf will cache metric_buffer_limit metrics for each
|
||||
## output, and will flush this buffer on a successful write. Oldest metrics
|
||||
## are dropped first when this buffer fills.
|
||||
## This buffer only fills when writes fail to output plugin(s).
|
||||
metric_buffer_limit = 10000
|
||||
|
||||
## Collection jitter is used to jitter the collection by a random amount.
|
||||
## Each plugin will sleep for a random time within jitter before collecting.
|
||||
## This can be used to avoid many plugins querying things like sysfs at the
|
||||
## same time, which can have a measurable effect on the system.
|
||||
collection_jitter = "0s"
|
||||
|
||||
## Default flushing interval for all outputs. Maximum flush_interval will be
|
||||
## flush_interval + flush_jitter
|
||||
flush_interval = "10s"
|
||||
## Jitter the flush interval by a random amount. This is primarily to avoid
|
||||
## large write spikes for users running a large number of telegraf instances.
|
||||
## ie, a jitter of 5s and interval 10s means flushes will happen every 10-15s
|
||||
flush_jitter = "0s"
|
||||
|
||||
## By default or when set to "0s", precision will be set to the same
|
||||
## timestamp order as the collection interval, with the maximum being 1s.
|
||||
## ie, when interval = "10s", precision will be "1s"
|
||||
## when interval = "250ms", precision will be "1ms"
|
||||
## Precision will NOT be used for service inputs. It is up to each individual
|
||||
## service input to set the timestamp at the appropriate precision.
|
||||
## Valid time units are "ns", "us" (or "µs"), "ms", "s".
|
||||
precision = ""
|
||||
|
||||
## Logging configuration:
|
||||
## Run telegraf with debug log messages.
|
||||
debug = false
|
||||
## Run telegraf in quiet mode (error log messages only).
|
||||
quiet = false
|
||||
## Specify the log file name. The empty string means to log to stderr.
|
||||
logfile = ""
|
||||
|
||||
## Override default hostname, if empty use os.Hostname()
|
||||
hostname = ""
|
||||
## If set to true, do no set the "host" tag in the telegraf agent.
|
||||
omit_hostname = false
|
||||
[[inputs.cpu]]
|
||||
## Whether to report per-cpu stats or not
|
||||
percpu = true
|
||||
## Whether to report total system cpu stats or not
|
||||
totalcpu = true
|
||||
## If true, collect raw CPU time metrics.
|
||||
collect_cpu_time = false
|
||||
## If true, compute and report the sum of all non-idle CPU states.
|
||||
report_active = false
|
||||
[[outputs.influxdb_v2]]
|
||||
## The URLs of the InfluxDB cluster nodes.
|
||||
##
|
||||
## Multiple URLs can be specified for a single cluster, only ONE of the
|
||||
## urls will be written to each interval.
|
||||
## urls exp: http://127.0.0.1:9999
|
||||
urls = ["http://127.0.0.1:9999"]
|
||||
|
||||
## Token for authentication.
|
||||
token = "no_more_secrets"
|
||||
|
||||
## Organization is the name of the organization you wish to write to; must exist.
|
||||
organization = "my_org"
|
||||
|
||||
## Destination bucket to write into.
|
||||
bucket = "my_bucket"
|
||||
`,
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
logger := zaptest.NewLogger(t)
|
||||
mapping := mock.NewUserResourceMappingService()
|
||||
labels := mock.NewLabelService()
|
||||
users := mock.NewUserService()
|
||||
|
||||
tt.r.Header.Set("Accept", tt.acceptHeader)
|
||||
w := httptest.NewRecorder()
|
||||
h := NewTelegrafHandler(logger, mapping, labels, tt.svc, users)
|
||||
|
||||
h.ServeHTTP(w, tt.r)
|
||||
|
||||
res := w.Result()
|
||||
content := res.Header.Get("Content-Type")
|
||||
body, _ := ioutil.ReadAll(res.Body)
|
||||
|
||||
if res.StatusCode != tt.wants.statusCode {
|
||||
t.Errorf("%q. handleGetTelegraf() = %v, want %v", tt.name, res.StatusCode, tt.wants.statusCode)
|
||||
t.Logf("headers: %v", res.Header)
|
||||
}
|
||||
if tt.wants.contentType != "" && content != tt.wants.contentType {
|
||||
t.Errorf("%q. handleGetTelegraf() = %v, want %v", tt.name, content, tt.wants.contentType)
|
||||
return
|
||||
}
|
||||
|
||||
if strings.Contains(tt.wants.contentType, "application/json") {
|
||||
if eq, _ := jsonEqual(string(body), tt.wants.body); tt.wants.body != "" && !eq {
|
||||
t.Errorf("%q. handleGetTelegraf() = \n***%v***\n,\nwant\n***%v***", tt.name, string(body), tt.wants.body)
|
||||
}
|
||||
} else if string(body) != tt.wants.body {
|
||||
t.Errorf("%q. handleGetTelegraf() = \n***%v***\n,\nwant\n***%v***", tt.name, string(body), tt.wants.body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_newTelegrafResponses(t *testing.T) {
|
||||
type args struct {
|
||||
tcs []*platform.TelegrafConfig
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
want string
|
||||
}{
|
||||
{
|
||||
args: args{
|
||||
tcs: []*platform.TelegrafConfig{
|
||||
&platform.TelegrafConfig{
|
||||
ID: platform.ID(1),
|
||||
Name: "my config",
|
||||
Agent: platform.TelegrafAgentConfig{
|
||||
Interval: 10000,
|
||||
},
|
||||
Plugins: []platform.TelegrafPlugin{
|
||||
{
|
||||
Comment: "my cpu stats",
|
||||
Config: &inputs.CPUStats{},
|
||||
},
|
||||
{
|
||||
Comment: "my influx output",
|
||||
Config: &outputs.InfluxDBV2{
|
||||
URLs: []string{"http://127.0.0.1:9999"},
|
||||
Token: "no_more_secrets",
|
||||
Organization: "my_org",
|
||||
Bucket: "my_bucket",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
want: `{
|
||||
"configurations": [
|
||||
{
|
||||
"id": "0000000000000001",
|
||||
"name": "my config",
|
||||
"agent": {
|
||||
"collectionInterval": 10000
|
||||
},
|
||||
"plugins": [
|
||||
{
|
||||
"name": "cpu",
|
||||
"type": "input",
|
||||
"comment": "my cpu stats",
|
||||
"config": {}
|
||||
},
|
||||
{
|
||||
"name": "influxdb_v2",
|
||||
"type": "output",
|
||||
"comment": "my influx output",
|
||||
"config": {
|
||||
"urls": [
|
||||
"http://127.0.0.1:9999"
|
||||
],
|
||||
"token": "no_more_secrets",
|
||||
"organization": "my_org",
|
||||
"bucket": "my_bucket"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}`,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
res := newTelegrafResponses(tt.args.tcs)
|
||||
got, err := json.Marshal(res)
|
||||
if err != nil {
|
||||
t.Fatalf("newTelegrafResponses() JSON marshal %v", err)
|
||||
}
|
||||
if eq, _ := jsonEqual(string(got), tt.want); tt.want != "" && !eq {
|
||||
t.Errorf("%q. newTelegrafResponses() = \n***%v***\n,\nwant\n***%v***", tt.name, string(got), tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_newTelegrafResponse(t *testing.T) {
|
||||
type args struct {
|
||||
tc *platform.TelegrafConfig
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
want string
|
||||
}{
|
||||
{
|
||||
args: args{
|
||||
tc: &platform.TelegrafConfig{
|
||||
ID: platform.ID(1),
|
||||
Name: "my config",
|
||||
Agent: platform.TelegrafAgentConfig{
|
||||
Interval: 10000,
|
||||
},
|
||||
Plugins: []platform.TelegrafPlugin{
|
||||
{
|
||||
Comment: "my cpu stats",
|
||||
Config: &inputs.CPUStats{},
|
||||
},
|
||||
{
|
||||
Comment: "my influx output",
|
||||
Config: &outputs.InfluxDBV2{
|
||||
URLs: []string{"http://127.0.0.1:9999"},
|
||||
Token: "no_more_secrets",
|
||||
Organization: "my_org",
|
||||
Bucket: "my_bucket",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
want: `{
|
||||
"id": "0000000000000001",
|
||||
"name": "my config",
|
||||
"agent": {
|
||||
"collectionInterval": 10000
|
||||
},
|
||||
"plugins": [
|
||||
{
|
||||
"name": "cpu",
|
||||
"type": "input",
|
||||
"comment": "my cpu stats",
|
||||
"config": {}
|
||||
},
|
||||
{
|
||||
"name": "influxdb_v2",
|
||||
"type": "output",
|
||||
"comment": "my influx output",
|
||||
"config": {
|
||||
"urls": [
|
||||
"http://127.0.0.1:9999"
|
||||
],
|
||||
"token": "no_more_secrets",
|
||||
"organization": "my_org",
|
||||
"bucket": "my_bucket"
|
||||
}
|
||||
}
|
||||
]
|
||||
}`,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
res := newTelegrafResponse(tt.args.tc)
|
||||
got, err := json.Marshal(res)
|
||||
if err != nil {
|
||||
t.Fatalf("newTelegrafResponse() JSON marshal %v", err)
|
||||
}
|
||||
if eq, _ := jsonEqual(string(got), tt.want); tt.want != "" && !eq {
|
||||
t.Errorf("%q. newTelegrafResponse() = \n%v\n,\nwant\n%v", tt.name, string(got), tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,69 @@
|
|||
package mock
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/influxdata/platform"
|
||||
)
|
||||
|
||||
var _ platform.TelegrafConfigStore = &TelegrafConfigStore{}
|
||||
|
||||
// TelegrafConfigStore represents a service for managing telegraf config data.
|
||||
type TelegrafConfigStore struct {
|
||||
FindUserResourceMappingsF func(ctx context.Context, filter platform.UserResourceMappingFilter, opt ...platform.FindOptions) ([]*platform.UserResourceMapping, int, error)
|
||||
CreateUserResourceMappingF func(ctx context.Context, m *platform.UserResourceMapping) error
|
||||
DeleteUserResourceMappingF func(ctx context.Context, resourceID platform.ID, userID platform.ID) error
|
||||
FindTelegrafConfigByIDF func(ctx context.Context, id platform.ID) (*platform.TelegrafConfig, error)
|
||||
FindTelegrafConfigF func(ctx context.Context, filter platform.UserResourceMappingFilter) (*platform.TelegrafConfig, error)
|
||||
FindTelegrafConfigsF func(ctx context.Context, filter platform.UserResourceMappingFilter, opt ...platform.FindOptions) ([]*platform.TelegrafConfig, int, error)
|
||||
CreateTelegrafConfigF func(ctx context.Context, tc *platform.TelegrafConfig, userID platform.ID) error
|
||||
UpdateTelegrafConfigF func(ctx context.Context, id platform.ID, tc *platform.TelegrafConfig, userID platform.ID) (*platform.TelegrafConfig, error)
|
||||
DeleteTelegrafConfigF func(ctx context.Context, id platform.ID) error
|
||||
}
|
||||
|
||||
// FindUserResourceMappings returns a list of UserResourceMappings that match filter and the total count of matching mappings.
|
||||
func (s *TelegrafConfigStore) FindUserResourceMappings(ctx context.Context, filter platform.UserResourceMappingFilter, opt ...platform.FindOptions) ([]*platform.UserResourceMapping, int, error) {
|
||||
return s.FindUserResourceMappingsF(ctx, filter, opt...)
|
||||
}
|
||||
|
||||
// CreateUserResourceMapping creates a user resource mapping
|
||||
func (s *TelegrafConfigStore) CreateUserResourceMapping(ctx context.Context, m *platform.UserResourceMapping) error {
|
||||
return s.CreateUserResourceMappingF(ctx, m)
|
||||
}
|
||||
|
||||
// DeleteUserResourceMapping deletes a user resource mapping
|
||||
func (s *TelegrafConfigStore) DeleteUserResourceMapping(ctx context.Context, resourceID platform.ID, userID platform.ID) error {
|
||||
return s.DeleteUserResourceMappingF(ctx, resourceID, userID)
|
||||
}
|
||||
|
||||
// FindTelegrafConfigByID returns a single telegraf config by ID.
|
||||
func (s *TelegrafConfigStore) FindTelegrafConfigByID(ctx context.Context, id platform.ID) (*platform.TelegrafConfig, error) {
|
||||
return s.FindTelegrafConfigByIDF(ctx, id)
|
||||
}
|
||||
|
||||
// FindTelegrafConfig returns the first telegraf config that matches filter.
|
||||
func (s *TelegrafConfigStore) FindTelegrafConfig(ctx context.Context, filter platform.UserResourceMappingFilter) (*platform.TelegrafConfig, error) {
|
||||
return s.FindTelegrafConfigF(ctx, filter)
|
||||
}
|
||||
|
||||
// FindTelegrafConfigs returns a list of telegraf configs that match filter and the total count of matching telegraf configs.
|
||||
// Additional options provide pagination & sorting.
|
||||
func (s *TelegrafConfigStore) FindTelegrafConfigs(ctx context.Context, filter platform.UserResourceMappingFilter, opt ...platform.FindOptions) ([]*platform.TelegrafConfig, int, error) {
|
||||
return s.FindTelegrafConfigsF(ctx, filter, opt...)
|
||||
}
|
||||
|
||||
// CreateTelegrafConfig creates a new telegraf config and sets b.ID with the new identifier.
|
||||
func (s *TelegrafConfigStore) CreateTelegrafConfig(ctx context.Context, tc *platform.TelegrafConfig, userID platform.ID) error {
|
||||
return s.CreateTelegrafConfigF(ctx, tc, userID)
|
||||
}
|
||||
|
||||
// UpdateTelegrafConfig updates a single telegraf config.
|
||||
// Returns the new telegraf config after update.
|
||||
func (s *TelegrafConfigStore) UpdateTelegrafConfig(ctx context.Context, id platform.ID, tc *platform.TelegrafConfig, userID platform.ID) (*platform.TelegrafConfig, error) {
|
||||
return s.UpdateTelegrafConfigF(ctx, id, tc, userID)
|
||||
}
|
||||
|
||||
// DeleteTelegrafConfig removes a telegraf config by ID.
|
||||
func (s *TelegrafConfigStore) DeleteTelegrafConfig(ctx context.Context, id platform.ID) error {
|
||||
return s.DeleteTelegrafConfigF(ctx, id)
|
||||
}
|
||||
Loading…
Reference in New Issue