diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b14688c6011..1aec27651b85 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ can be found at [#317](https://github.com/grafana/agent/issues/317). # Master (unreleased) +- [FEATURE] Prometheus `remote_write` now supports SigV4 authentication using + the [AWS default credentials + chain](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html). + This enables the Agent to send metrics to Amazon Managed Prometheus without + needing the [SigV4 Proxy](https://github.com/awslabs/aws-sigv4-proxy). + (@rfratto) + - [ENHANCEMENT] Update `redis_exporter` to v1.15.0. (@rfratto) - [ENHANCEMENT] `memcached_exporter` has been updated to v0.8.0. (@rfratto) diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md index a39976778e72..d737a454c7d7 100644 --- a/docs/configuration-reference.md +++ b/docs/configuration-reference.md @@ -1812,6 +1812,25 @@ basic_auth: # read from the configured file. It is mutually exclusive with `bearer_token`. [ bearer_token_file: /path/to/bearer/token/file ] +# Configures SigV4 request signing. The default credentials chain will be used, +# documented here: +# +# https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html#specifying-credentials +# +# When enabled, region must be supplied. +# +# This feature is currently exclusive to the Grafana Cloud Agent and is only +# currently available for remote_write of Prometheus metrics. +sigv4: + # Enable SigV4 request signing. May not be enabled at the same time as + # configuring basic auth or bearer_token/bearer_token_file. + [ enabled: | default = false ] + + # Region to use for signing the requests. When sigv4.enabled is true, + # must be non-empty and must be the region of the AMP workspace specified + # by the remote_write URL. + region: + # Configures the remote write request's TLS settings. tls_config: [ ] diff --git a/go.mod b/go.mod index 38e00e4712b3..69b8718380d1 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.12 require ( contrib.go.opencensus.io/exporter/prometheus v0.2.0 + github.com/aws/aws-sdk-go v1.35.5 github.com/cortexproject/cortex v1.4.1-0.20201022071705-85942c5703cf github.com/drone/envsubst v1.0.2 github.com/go-kit/kit v0.10.0 @@ -69,7 +70,7 @@ replace ( github.com/satori/go.uuid => github.com/satori/go.uuid v1.2.0 ) -replace github.com/prometheus/prometheus => github.com/grafana/prometheus v1.8.2-0.20201021200247-cf00050ed1e9 +replace github.com/prometheus/prometheus => github.com/grafana/prometheus v1.8.2-0.20210111220521-c0d5de2f0ee3 replace gopkg.in/yaml.v2 => github.com/rfratto/go-yaml v0.0.0-20200521142311-984fc90c8a04 diff --git a/go.sum b/go.sum index f8023f50fd5f..babba0a53718 100644 --- a/go.sum +++ b/go.sum @@ -998,6 +998,8 @@ github.com/grafana/prometheus v1.8.2-0.20200821135656-2efe42db3b77 h1:BDISe7QQ7y github.com/grafana/prometheus v1.8.2-0.20200821135656-2efe42db3b77/go.mod h1:nnrpCyzNlHMAvHQl3Edz6cUiJZu3q4owFNV3oU3K7x0= github.com/grafana/prometheus v1.8.2-0.20201021200247-cf00050ed1e9 h1:pYjsijDKlGYurpki3DAdAgAN7MMXDmBdKjhtyqusr/E= github.com/grafana/prometheus v1.8.2-0.20201021200247-cf00050ed1e9/go.mod h1:Td6hjwdXDmVt5CI9T03Sw+yBNxLBq/Yx3ZtmtP8zlCA= +github.com/grafana/prometheus v1.8.2-0.20210111220521-c0d5de2f0ee3 h1:8D1tqy2auT6U5/OdEy0zutgj9Jr8SNlOzPAP1JnfLM8= +github.com/grafana/prometheus v1.8.2-0.20210111220521-c0d5de2f0ee3/go.mod h1:Td6hjwdXDmVt5CI9T03Sw+yBNxLBq/Yx3ZtmtP8zlCA= github.com/grafana/redis_exporter v1.13.0 h1:PvcKv41q+z4bY4iglw0M9cJVn3Qxq5L2mWcc96pZqPE= github.com/grafana/redis_exporter v1.13.0/go.mod h1:yJS6eupd/mUMsrcwIyG9avqSDPsAKrtB3enG5ghYXi8= github.com/grafana/statsd_exporter v0.18.1-0.20201120191414-b5deeda251f5 h1:xxinXFunwgA3p3ciVPEYHFEjgFsxFXPUb0aEquwcwgk= @@ -2533,6 +2535,7 @@ golang.org/x/tools v0.0.0-20201022035929-9cf592e881e9/go.mod h1:emZCQorbCU4vsT4f golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gonum.org/v1/gonum v0.0.0-20180816165407-929014505bf4/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo= gonum.org/v1/gonum v0.0.0-20181121035319-3f7ecaa7e8ca/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo= @@ -2592,6 +2595,7 @@ google.golang.org/genproto v0.0.0-20200904004341-0bd0a958aa1d h1:92D1fum1bJLKSdr google.golang.org/genproto v0.0.0-20200904004341-0bd0a958aa1d/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no= google.golang.org/grpc v1.29.1 h1:EC2SB8S04d2r73uptxphDSUG+kTKVgjRPF+N3xpxRB4= google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3IjizoKk= +google.golang.org/grpc/examples v0.0.0-20200728065043-dfc0c05b2da9 h1:f+/+gfZ/tfaHBXXiv1gWRmCej6wlX3mLY4bnLpI99wk= google.golang.org/grpc/examples v0.0.0-20200728065043-dfc0c05b2da9/go.mod h1:5j1uub0jRGhRiSghIlrThmBUgcgLXOVJQ/l1getT4uo= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= @@ -2609,6 +2613,7 @@ gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLks gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b h1:QRR6H1YWRnHb4Y/HeNFCTJLFVxaq6wH4YuVdsUOr75U= gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/cheggaaa/pb.v1 v1.0.25/go.mod h1:V/YB90LKu/1FcN3WVnfiiE5oMCibMjukxqG/qStrOgw= gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI= @@ -2624,11 +2629,16 @@ gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/ini.v1 v1.52.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/ini.v1 v1.57.0 h1:9unxIsFcTt4I55uWluz+UmL95q4kdJ0buvQ1ZIqVQww= gopkg.in/ini.v1 v1.57.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= +gopkg.in/jcmturner/aescts.v1 v1.0.1 h1:cVVZBK2b1zY26haWB4vbBiZrfFQnfbTVrE3xZq6hrEw= gopkg.in/jcmturner/aescts.v1 v1.0.1/go.mod h1:nsR8qBOg+OucoIW+WMhB3GspUQXq9XorLnQb9XtvcOo= +gopkg.in/jcmturner/dnsutils.v1 v1.0.1 h1:cIuC1OLRGZrld+16ZJvvZxVJeKPsvd5eUIvxfoN5hSM= gopkg.in/jcmturner/dnsutils.v1 v1.0.1/go.mod h1:m3v+5svpVOhtFAP/wSz+yzh4Mc0Fg7eRhxkJMWSIz9Q= +gopkg.in/jcmturner/goidentity.v3 v3.0.0 h1:1duIyWiTaYvVx3YX2CYtpJbUFd7/UuPYCfgXtQ3VTbI= gopkg.in/jcmturner/goidentity.v3 v3.0.0/go.mod h1:oG2kH0IvSYNIu80dVAyu/yoefjq1mNfM5bm88whjWx4= gopkg.in/jcmturner/gokrb5.v7 v7.2.3/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM= +gopkg.in/jcmturner/gokrb5.v7 v7.5.0 h1:a9tsXlIDD9SKxotJMK3niV7rPZAJeX2aD/0yg3qlIrg= gopkg.in/jcmturner/gokrb5.v7 v7.5.0/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM= +gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU= gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8= gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo= gopkg.in/square/go-jose.v2 v2.5.1 h1:7odma5RETjNHWJnR32wx8t+Io4djHE1PqxCFx3iiZ2w= @@ -2639,7 +2649,9 @@ gopkg.in/warnings.v0 v0.1.2/go.mod h1:jksf8JmL6Qr/oQM2OXTHunEvvTAsrWBLb6OOjuVWRN gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 h1:tQIYjPdBoyREyB9XMu+nnTclpTYkz2zFM+lzLJFO4gQ= gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo= gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw= +gotest.tools/v3 v3.0.2 h1:kG1BFyqVHuQoVQiR1bWGnfz/fmHvvuiSPIV7rvl360E= gotest.tools/v3 v3.0.2/go.mod h1:3SzNCllyD9/Y+b5r9JIKQ474KzkZyqLqEfYqMsX94Bk= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= @@ -2648,6 +2660,7 @@ honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWh honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg= honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k= honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k= +honnef.co/go/tools v0.0.1-2020.1.6 h1:W18jzjh8mfPez+AwGLxmOImucz/IFjpNlrKVnaj2YVc= honnef.co/go/tools v0.0.1-2020.1.6/go.mod h1:pyyisuGw24ruLjrr1ddx39WE0y9OooInRzEYLhQB2YY= howett.net/plist v0.0.0-20181124034731-591f970eefbb/go.mod h1:vMygbs4qMhSZSc4lCUl2OEE+rDiIIJAIdR4m7MiMcm0= k8s.io/api v0.0.0-20191115095533-47f6de673b26/go.mod h1:iA/8arsvelvo4IDqIhX4IbjTEKBGgvsf2OraTuRtLFU= @@ -2664,6 +2677,7 @@ k8s.io/gengo v0.0.0-20190128074634-0689ccc1d7d6/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8 k8s.io/klog/v2 v2.0.0 h1:Foj74zO6RbjjP4hBEKjnYtjjAhGg4jNynUdYF6fJrok= k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE= k8s.io/kube-openapi v0.0.0-20191107075043-30be4d16710a/go.mod h1:1TqjTSzOxsLGIKfj0lK8EeCP7K1iUG65v09OM0/WG5E= +k8s.io/kube-openapi v0.0.0-20200410145947-61e04a5be9a6 h1:Oh3Mzx5pJ+yIumsAD0MOECPVeXsVot0UkiaCGVyfGQY= k8s.io/kube-openapi v0.0.0-20200410145947-61e04a5be9a6/go.mod h1:GRQhZsXIAJ1xR0C9bd8UpWHZ5plfAS9fzPjJuQ6JL3E= k8s.io/utils v0.0.0-20191114200735-6ca3b61696b6/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew= k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew= diff --git a/pkg/integrations/manager.go b/pkg/integrations/manager.go index 1df71561e670..e246b661f5c2 100644 --- a/pkg/integrations/manager.go +++ b/pkg/integrations/manager.go @@ -57,7 +57,7 @@ type ManagerConfig struct { Labels model.LabelSet `yaml:"labels,omitempty"` // Prometheus RW configs to use for all integrations. - PrometheusRemoteWrite []*config.RemoteWriteConfig `yaml:"prometheus_remote_write,omitempty"` + PrometheusRemoteWrite []*instance.RemoteWriteConfig `yaml:"prometheus_remote_write,omitempty"` IntegrationRestartBackoff time.Duration `yaml:"integration_restart_backoff,omitempty"` diff --git a/pkg/prom/instance/group_manager.go b/pkg/prom/instance/group_manager.go index 9befb9a55eff..1e70ccf4db4c 100644 --- a/pkg/prom/instance/group_manager.go +++ b/pkg/prom/instance/group_manager.go @@ -247,7 +247,7 @@ func hashConfig(c Config) (string, error) { // Assign names to remote_write configs if they're not present already. // This is also done in AssignDefaults but is duplicated here for the sake // of simplifying responsibility of GroupManager. - for _, cfg := range groupable.RemoteWrite { + for _, cfg := range groupable.BaseRemoteWrite() { if cfg != nil { // We don't care if the names are different, just that the other settings // are the same. Blank out the name here before hashing the remote @@ -270,7 +270,7 @@ func hashConfig(c Config) (string, error) { case groupable.RemoteWrite[j] == nil: return false default: - return groupable.RemoteWrite[i].Name < groupable.RemoteWrite[j].Name + return groupable.RemoteWrite[i].Base.Name < groupable.RemoteWrite[j].Base.Name } }) @@ -301,7 +301,7 @@ func copyConfig(c Config) (Config, error) { cfg.ScrapeConfigs = []*config.ScrapeConfig{} } if cfg.RemoteWrite == nil && c.RemoteWrite != nil { - cfg.RemoteWrite = []*config.RemoteWriteConfig{} + cfg.RemoteWrite = []*RemoteWriteConfig{} } return *cfg, nil } @@ -334,14 +334,14 @@ func groupConfigs(groupName string, grouped groupedConfigs) (Config, error) { for _, rwc := range combined.RemoteWrite { // Blank out the existing name before getting the hash so it is doesn't take into // account any existing name. - rwc.Name = "" + rwc.Base.Name = "" hash, err := getHash(rwc) if err != nil { return Config{}, err } - rwc.Name = groupName[:6] + "-" + hash[:6] + rwc.Base.Name = groupName[:6] + "-" + hash[:6] } // Combine all the scrape configs. It's possible that two different ungrouped diff --git a/pkg/prom/instance/group_manager_test.go b/pkg/prom/instance/group_manager_test.go index 48b87236f4b0..6efab303b9af 100644 --- a/pkg/prom/instance/group_manager_test.go +++ b/pkg/prom/instance/group_manager_test.go @@ -193,7 +193,7 @@ remote_write: require.Equal(t, 1, len(innerConfigs)) cfg := innerConfigs[gm.groupLookup["configA"]] - require.NotEqual(t, "rw-cfg-a", cfg.RemoteWrite[0].Name) + require.NotEqual(t, "rw-cfg-a", cfg.RemoteWrite[0].Base.Name) } func TestGroupManager_DeleteConfig(t *testing.T) { @@ -423,7 +423,7 @@ remote_write: for _, rwConfig := range expect.RemoteWrite { hash, err := getHash(rwConfig) require.NoError(t, err) - rwConfig.Name = groupName[:6] + "-" + hash[:6] + rwConfig.Base.Name = groupName[:6] + "-" + hash[:6] } group := groupedConfigs{ diff --git a/pkg/prom/instance/instance.go b/pkg/prom/instance/instance.go index fd09d971a68f..36cf92f13a8a 100644 --- a/pkg/prom/instance/instance.go +++ b/pkg/prom/instance/instance.go @@ -55,11 +55,11 @@ var ( // Config is a specific agent that runs within the overall Prometheus // agent. It has its own set of scrape_configs and remote_write rules. type Config struct { - Name string `yaml:"name" json:"name"` - HostFilter bool `yaml:"host_filter" json:"host_filter"` - HostFilterRelabelConfigs []*relabel.Config `yaml:"host_filter_relabel_configs,omitempty"` - ScrapeConfigs []*config.ScrapeConfig `yaml:"scrape_configs,omitempty" json:"scrape_configs,omitempty"` - RemoteWrite []*config.RemoteWriteConfig `yaml:"remote_write,omitempty" json:"remote_write,omitempty"` + Name string `yaml:"name" json:"name"` + HostFilter bool `yaml:"host_filter" json:"host_filter"` + HostFilterRelabelConfigs []*relabel.Config `yaml:"host_filter_relabel_configs,omitempty"` + ScrapeConfigs []*config.ScrapeConfig `yaml:"scrape_configs,omitempty" json:"scrape_configs,omitempty"` + RemoteWrite []*RemoteWriteConfig `yaml:"remote_write,omitempty" json:"remote_write,omitempty"` // How frequently the WAL should be truncated. WALTruncateFrequency time.Duration `yaml:"wal_truncate_frequency,omitempty" json:"wal_truncate_frequency,omitempty"` @@ -72,6 +72,19 @@ type Config struct { WriteStaleOnShutdown bool `yaml:"write_stale_on_shutdown,omitempty" json:"write_stale_on_shutdown,omitempty"` } +// BaseRemoteWrite returns the base remote write configs without the added +// fields. +func (c *Config) BaseRemoteWrite() []*config.RemoteWriteConfig { + res := make([]*config.RemoteWriteConfig, len(c.RemoteWrite)) + for i, cfg := range c.RemoteWrite { + if cfg == nil { + continue + } + res[i] = &cfg.Base + } + return res +} + func (c *Config) UnmarshalYAML(unmarshal func(interface{}) error) error { *c = DefaultConfig @@ -153,7 +166,7 @@ func (c *Config) ApplyDefaults(global *config.GlobalConfig) error { // unique name to the config so we can pull metrics from it when running // an instance. var generatedName bool - if cfg.Name == "" { + if cfg.Base.Name == "" { hash, err := getHash(cfg) if err != nil { return err @@ -162,17 +175,17 @@ func (c *Config) ApplyDefaults(global *config.GlobalConfig) error { // We have to add the name of the instance to ensure that generated metrics // are unique across multiple agent instances. The remote write queues currently // globally register their metrics so we can't inject labels here. - cfg.Name = c.Name + "-" + hash[:6] + cfg.Base.Name = c.Name + "-" + hash[:6] generatedName = true } - if _, exists := rwNames[cfg.Name]; exists { + if _, exists := rwNames[cfg.Base.Name]; exists { if generatedName { return fmt.Errorf("found two identical remote_write configs") } - return fmt.Errorf("found duplicate remote write configs with name %q", cfg.Name) + return fmt.Errorf("found duplicate remote write configs with name %q", cfg.Base.Name) } - rwNames[cfg.Name] = struct{}{} + rwNames[cfg.Base.Name] = struct{}{} } return nil @@ -365,9 +378,10 @@ func (i *Instance) initialize(ctx context.Context, reg prometheus.Registerer, cf // Setup the remote storage remoteLogger := log.With(i.logger, "component", "remote") i.remoteStore = remote.NewStorage(remoteLogger, reg, i.wal.StartTime, i.wal.Directory(), cfg.RemoteFlushDeadline, i.readyScrapeManager) + i.remoteStore.Write.NewClient = i.newWriteClient err = i.remoteStore.ApplyConfig(&config.Config{ GlobalConfig: i.globalCfg, - RemoteWriteConfigs: cfg.RemoteWrite, + RemoteWriteConfigs: cfg.BaseRemoteWrite(), }) if err != nil { return fmt.Errorf("failed applying config to remote storage: %w", err) @@ -389,16 +403,48 @@ func (i *Instance) initialize(ctx context.Context, reg prometheus.Registerer, cf return nil } +func (i *Instance) newWriteClient(name string, conf *remote.ClientConfig) (remote.WriteClient, error) { + var cfg *RemoteWriteConfig + for _, c := range i.cfg.RemoteWrite { + if c.Base.Name == name { + cfg = c + break + } + } + if cfg == nil { + level.Warn(i.logger).Log("msg", "could not properly generate HTTP client for remote_write", "err", "could not find remote write config", "name", name) + } + + writeClient, err := remote.NewWriteClient(name, conf) + if err != nil { + return nil, err + } + // Based on NewWriteClient, this will always be a remote.Client. + cli := writeClient.(*remote.Client) + + if cfg != nil && cfg.SigV4.Enabled { + level.Debug(i.logger).Log("msg", "enabling sigv4", "name", name) + + rt, err := NewSigV4RoundTripper(cfg.SigV4, cli.Client.Transport) + if err != nil { + level.Error(i.logger).Log("msg", "failed to create sigv4 transport", "err", err) + return nil, fmt.Errorf("could not create sigv4 transport: %w", err) + } + cli.Client.Transport = rt + } + + return cli, nil +} + // Update accepts a new Config for the Instance and will dynamically update any // running Prometheus components with the new values from Config. Update will // return an ErrInvalidUpdate if the Update could not be applied. -func (i *Instance) Update(c Config) error { +func (i *Instance) Update(c Config) (err error) { i.mut.Lock() defer i.mut.Unlock() // It's only (currently) valid to update scrape_configs and remote_write, so // if any other field has changed here, return the error. - var err error switch { // This first case should never happen in practice but it's included here for // completions sake. @@ -429,13 +475,22 @@ func (i *Instance) Update(c Config) error { // // Keep the following order below: // - // 1. Remote Store - // 2. Scrape Manager - // 3. Discovery Manager + // 1. Local config + // 2. Remote Store + // 3. Scrape Manager + // 4. Discovery Manager + + originalConfig := i.cfg + defer func() { + if err != nil { + i.cfg = originalConfig + } + }() + i.cfg = c err = i.remoteStore.ApplyConfig(&config.Config{ GlobalConfig: i.globalCfg, - RemoteWriteConfigs: c.RemoteWrite, + RemoteWriteConfigs: c.BaseRemoteWrite(), }) if err != nil { return fmt.Errorf("error applying new remote_write configs: %w", err) @@ -462,7 +517,6 @@ func (i *Instance) Update(c Config) error { return fmt.Errorf("failed applying configs to discovery manager: %w", err) } - i.cfg = c return nil } @@ -624,7 +678,7 @@ func (i *Instance) getRemoteWriteTimestamp() int64 { lbls := make([]string, len(i.cfg.RemoteWrite)) for idx := 0; idx < len(lbls); idx++ { - lbls[idx] = i.cfg.RemoteWrite[idx].Name + lbls[idx] = i.cfg.RemoteWrite[idx].Base.Name } vals, err := i.vc.GetValues("remote_name", lbls...) diff --git a/pkg/prom/instance/instance_integration_test.go b/pkg/prom/instance/instance_integration_test.go index 274c20ccf86c..b0ca1275c892 100644 --- a/pkg/prom/instance/instance_integration_test.go +++ b/pkg/prom/instance/instance_integration_test.go @@ -105,6 +105,70 @@ remote_write: }) } +func TestInstance_Update_Failed(t *testing.T) { + logger := log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr)) + + walDir, err := ioutil.TempDir(os.TempDir(), "wal") + require.NoError(t, err) + t.Cleanup(func() { os.RemoveAll(walDir) }) + + r := mux.NewRouter() + r.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { + promhttp.Handler().ServeHTTP(w, r) + }) + r.HandleFunc("/push", func(w http.ResponseWriter, r *http.Request) {}) + + // Start a server for exposing the router. + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer l.Close() + go func() { + _ = http.Serve(l, r) + }() + + // Create a new instance where it's not scraping or writing anything by default. + initialConfig := loadConfig(t, ` +name: integration_test +scrape_configs: [] +remote_write: [] +`) + inst, err := New(prometheus.NewRegistry(), config.DefaultGlobalConfig, initialConfig, walDir, logger) + require.NoError(t, err) + + instCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + err := inst.Run(instCtx) + require.NoError(t, err) + }() + + // Create a new config to use for updating + newConfig := loadConfig(t, fmt.Sprintf(` +name: integration_test +scrape_configs: + - job_name: test_scrape + scrape_interval: 5s + static_configs: + - targets: ['%[1]s'] +remote_write: + - url: http://%[1]s/push +`, l.Addr())) + + // Make sure the instance can successfully update first + test.Poll(t, time.Second*15, nil, func() interface{} { + err := inst.Update(newConfig) + if err != nil { + logger.Log("msg", "failed to update instance", "err", err) + } + return err + }) + + // Now force an update back to the original config to fail + inst.readyScrapeManager.Set(nil) + require.NotNil(t, inst.Update(initialConfig), "update should have failed") + require.Equal(t, newConfig, inst.cfg, "config did not roll back") +} + // TestInstance_Update_InvalidChanges runs an instance with a blank initial // config and performs various unacceptable updates that should return an // error. diff --git a/pkg/prom/instance/instance_test.go b/pkg/prom/instance/instance_test.go index c9b13a48980d..f43145577abb 100644 --- a/pkg/prom/instance/instance_test.go +++ b/pkg/prom/instance/instance_test.go @@ -68,8 +68,8 @@ func TestConfig_ApplyDefaults_Validations(t *testing.T) { }}, }, }} - cfg.RemoteWrite = []*config.RemoteWriteConfig{{ - Name: "write", + cfg.RemoteWrite = []*RemoteWriteConfig{{ + Base: config.RemoteWriteConfig{Name: "write"}, }} tt := []struct { @@ -130,8 +130,8 @@ func TestConfig_ApplyDefaults_Validations(t *testing.T) { "multiple remote writes with same name", func(c *Config) { c.RemoteWrite = append(c.RemoteWrite, - &config.RemoteWriteConfig{Name: "foo"}, - &config.RemoteWriteConfig{Name: "foo"}, + &RemoteWriteConfig{Base: config.RemoteWriteConfig{Name: "foo"}}, + &RemoteWriteConfig{Base: config.RemoteWriteConfig{Name: "foo"}}, ) }, fmt.Errorf("found duplicate remote write configs with name \"foo\""), @@ -151,7 +151,7 @@ func TestConfig_ApplyDefaults_Validations(t *testing.T) { } input.ScrapeConfigs = scrapeConfigs - var remoteWrites []*config.RemoteWriteConfig + var remoteWrites []*RemoteWriteConfig for _, rw := range input.RemoteWrite { rwCopy := *rw remoteWrites = append(remoteWrites, &rwCopy) @@ -172,6 +172,23 @@ func TestConfig_ApplyDefaults_Validations(t *testing.T) { } } +func TestConfig_ApplyDefaults_HashedName(t *testing.T) { + global := config.DefaultGlobalConfig + + cfgText := ` +name: default +host_filter: false +remote_write: +- url: http://localhost:9009/api/prom/push + sigv4: + enabled: true` + + cfg, err := UnmarshalConfig(strings.NewReader(cfgText)) + require.NoError(t, err) + require.NoError(t, cfg.ApplyDefaults(&global)) + require.NotEmpty(t, cfg.RemoteWrite[0].Base.Name) +} + func TestInstance_Path(t *testing.T) { scrapeAddr, closeSrv := getTestServer(t) defer closeSrv() diff --git a/pkg/prom/instance/marshal_test.go b/pkg/prom/instance/marshal_test.go index 1489e140cf16..f0fa89e698e5 100644 --- a/pkg/prom/instance/marshal_test.go +++ b/pkg/prom/instance/marshal_test.go @@ -96,3 +96,74 @@ remote_flush_deadline: 1m0s require.YAMLEq(t, cfg, string(out)) }) } + +// TestMarshal_UnmarshalConfig ensures that any method of marshaling an +// instance config does the same thing and retains secrets. +func TestMarshal_UnmarshalConfig_Sigv4(t *testing.T) { + cfg := `name: test +host_filter: false +scrape_configs: +- job_name: local_scrape + honor_timestamps: true + metrics_path: /metrics + scheme: http + static_configs: + - targets: + - 127.0.0.1:12345 + labels: + cluster: localhost + basic_auth: + username: admin + password: foobar +remote_write: +- url: http://localhost:9009/api/prom/push + remote_timeout: 30s + name: test-d0f32c + sigv4: + enabled: true + queue_config: + capacity: 500 + max_shards: 1000 + min_shards: 1 + max_samples_per_send: 100 + batch_send_deadline: 5s + min_backoff: 30ms + max_backoff: 100ms + metadata_config: + send: true + send_interval: 1m +wal_truncate_frequency: 1m0s +min_wal_time: 5m0s +max_wal_time: 4h0m0s +remote_flush_deadline: 1m0s +` + + t.Run("direct marshal", func(t *testing.T) { + var c Config + err := yaml.Unmarshal([]byte(cfg), &c) + require.NoError(t, err) + + out, err := yaml.Marshal(c) + require.NoError(t, err) + require.YAMLEq(t, cfg, string(out)) + }) + + t.Run("direct mashal pointer", func(t *testing.T) { + c := &Config{} + err := yaml.Unmarshal([]byte(cfg), c) + require.NoError(t, err) + + out, err := yaml.Marshal(c) + require.NoError(t, err) + require.YAMLEq(t, cfg, string(out)) + }) + + t.Run("custom marshal methods", func(t *testing.T) { + c, err := UnmarshalConfig(strings.NewReader(cfg)) + require.NoError(t, err) + + out, err := MarshalConfig(c, false) + require.NoError(t, err) + require.YAMLEq(t, cfg, string(out)) + }) +} diff --git a/pkg/prom/instance/remote_write.go b/pkg/prom/instance/remote_write.go new file mode 100644 index 000000000000..a98a27824a17 --- /dev/null +++ b/pkg/prom/instance/remote_write.go @@ -0,0 +1,69 @@ +package instance + +import ( + "fmt" + + "github.com/prometheus/prometheus/config" +) + +// RemoteWriteConfig extends the default RemoteWriteConfig with extra settings. +type RemoteWriteConfig struct { + Base config.RemoteWriteConfig `yaml:",inline"` + + SigV4 SigV4Config `yaml:"sigv4,omitempty"` +} + +func (c *RemoteWriteConfig) UnmarshalYAML(unmarshal func(interface{}) error) error { + c.Base = config.DefaultRemoteWriteConfig + + type plain RemoteWriteConfig + if err := unmarshal((*plain)(c)); err != nil { + return err + } + + // NOTE(rfratto): UnmarshalYAML can't be called for inlined fields, which + // must not be pointers. We need to copy the validation logic here and + // sync changes when they happen upstream. + if c.Base.URL == nil { + return fmt.Errorf("url for remote_write is empty") + } + for _, cfg := range c.Base.WriteRelabelConfigs { + if cfg == nil { + return fmt.Errorf("empty or null relabeling rule in remote write config") + } + } + + return c.Validate() +} + +// Validate validates the HTTPClientConfig along with the SigV4Config to ensure only one +// authentication mechanism is used. +func (c *RemoteWriteConfig) Validate() error { + clientConfig := c.Base.HTTPClientConfig + if err := clientConfig.Validate(); err != nil { + return err + } + + // count converts a true value to 1, allowing to sum truth conditions + // together to calculate how many conditions were true. + count := func(b bool) int { + if b { + return 1 + } + return 0 + } + + // Ensure at most one auth mechanism is enabled + var ( + usingBearer = count(len(clientConfig.BearerToken) > 0 || len(clientConfig.BearerTokenFile) > 0) + usingBasic = count(clientConfig.BasicAuth != nil) + usingSigV4 = count(c.SigV4.Enabled) + + enabled = usingBearer + usingBasic + usingSigV4 + ) + if enabled > 1 { + return fmt.Errorf("at most one of sigv4, basic auth, bearer tokens must be configured") + } + + return nil +} diff --git a/pkg/prom/instance/sigv4.go b/pkg/prom/instance/sigv4.go new file mode 100644 index 000000000000..12fd3cb84ac0 --- /dev/null +++ b/pkg/prom/instance/sigv4.go @@ -0,0 +1,96 @@ +package instance + +import ( + "bytes" + "fmt" + "io" + "io/ioutil" + "net/http" + "sync" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/session" + signer "github.com/aws/aws-sdk-go/aws/signer/v4" +) + +// SigV4Config configures signing requests with SigV4. +type SigV4Config struct { + Enabled bool `yaml:"enabled,omitempty"` + Region string `yaml:"region,omitempty"` +} + +type sigV4RoundTripper struct { + cfg SigV4Config + next http.RoundTripper + pool sync.Pool + + signer *signer.Signer +} + +// NewSigV4RoundTripper returns a new http.RoundTripper that will sign requests +// using Amazon's Signature Verification V4 signing procedure. The request will +// then be handed off to the next RoundTripper provided by next. If next is nil, +// http.DefaultTransport will be used. +// +// Credentials for signing are retrieving used the default AWS credential chain. +// If credentials could not be found, an error will be returned. +func NewSigV4RoundTripper(cfg SigV4Config, next http.RoundTripper) (http.RoundTripper, error) { + if cfg.Region == "" { + return nil, fmt.Errorf("region not configured") + } + + if next == nil { + next = http.DefaultTransport + } + + sess, err := session.NewSession(&aws.Config{ + Region: aws.String(cfg.Region), + }) + if err != nil { + return nil, fmt.Errorf("could not create new AWS session: %w", err) + } + if _, err := sess.Config.Credentials.Get(); err != nil { + return nil, fmt.Errorf("could not get sigv4 credentials: %w", err) + } + + rt := &sigV4RoundTripper{ + cfg: cfg, + next: next, + signer: signer.NewSigner(sess.Config.Credentials), + } + rt.pool.New = rt.newBuf + return rt, nil +} + +func (rt *sigV4RoundTripper) newBuf() interface{} { + return bytes.NewBuffer(make([]byte, 0, 1024)) +} + +func (rt *sigV4RoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + // rt.signer.Sign needs a seekable body, so we replace the body with a + // buffered reader filled with the contents of original body. + buf := rt.pool.Get().(*bytes.Buffer) + defer func() { + buf.Reset() + rt.pool.Put(buf) + }() + if _, err := io.Copy(buf, req.Body); err != nil { + return nil, err + } + // Close the original body since we don't need it anymore. + _ = req.Body.Close() + + // Ensure our seeker is back at the start of the buffer once we return. + var seeker io.ReadSeeker = bytes.NewReader(buf.Bytes()) + defer func() { + _, _ = seeker.Seek(0, io.SeekStart) + }() + req.Body = ioutil.NopCloser(seeker) + + _, err := rt.signer.Sign(req, seeker, "aps", rt.cfg.Region, time.Now().UTC()) + if err != nil { + return nil, fmt.Errorf("failed to sign request: %w", err) + } + return rt.next.RoundTrip(req) +} diff --git a/pkg/prom/instance/sigv4_test.go b/pkg/prom/instance/sigv4_test.go new file mode 100644 index 000000000000..da470d6e6adc --- /dev/null +++ b/pkg/prom/instance/sigv4_test.go @@ -0,0 +1,43 @@ +package instance + +import ( + "net/http" + "strings" + "testing" + + "github.com/aws/aws-sdk-go/aws/credentials" + signer "github.com/aws/aws-sdk-go/aws/signer/v4" + "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/stretchr/testify/require" +) + +func TestSigV4RoundTripper(t *testing.T) { + var gotReq *http.Request + + rt := &sigV4RoundTripper{ + cfg: SigV4Config{ + Enabled: true, + Region: "us-east-2", + }, + next: promhttp.RoundTripperFunc(func(req *http.Request) (*http.Response, error) { + gotReq = req + return &http.Response{StatusCode: http.StatusOK}, nil + }), + signer: signer.NewSigner(credentials.NewStaticCredentials( + "test-id", + "secret", + "token", + )), + } + rt.pool.New = rt.newBuf + + cli := &http.Client{Transport: rt} + + req, err := http.NewRequest(http.MethodPost, "google.com", strings.NewReader("Hello, world!")) + require.NoError(t, err) + _, err = cli.Do(req) + require.NoError(t, err) + + require.NotNil(t, gotReq) + require.NotEmpty(t, gotReq.Header.Get("Authorization")) +} diff --git a/vendor/github.com/prometheus/prometheus/storage/remote/storage.go b/vendor/github.com/prometheus/prometheus/storage/remote/storage.go index 65d4a61eb374..6c6bebd590f1 100644 --- a/vendor/github.com/prometheus/prometheus/storage/remote/storage.go +++ b/vendor/github.com/prometheus/prometheus/storage/remote/storage.go @@ -50,7 +50,7 @@ type Storage struct { logger log.Logger mtx sync.Mutex - rws *WriteStorage + Write *WriteStorage // For reads. queryables []storage.SampleAndChunkQueryable @@ -67,7 +67,7 @@ func NewStorage(l log.Logger, reg prometheus.Registerer, stCallback startTimeCal logger: logging.Dedupe(l, 1*time.Minute), localStartTimeCallback: stCallback, } - s.rws = NewWriteStorage(s.logger, reg, walDir, flushDeadline, sm) + s.Write = NewWriteStorage(s.logger, reg, walDir, flushDeadline, sm) return s } @@ -76,7 +76,7 @@ func (s *Storage) ApplyConfig(conf *config.Config) error { s.mtx.Lock() defer s.mtx.Unlock() - if err := s.rws.ApplyConfig(conf); err != nil { + if err := s.Write.ApplyConfig(conf); err != nil { return err } @@ -171,14 +171,14 @@ func (s *Storage) ChunkQuerier(ctx context.Context, mint, maxt int64) (storage.C // Appender implements storage.Storage. func (s *Storage) Appender(ctx context.Context) storage.Appender { - return s.rws.Appender(ctx) + return s.Write.Appender(ctx) } // Close the background processing of the storage queues. func (s *Storage) Close() error { s.mtx.Lock() defer s.mtx.Unlock() - return s.rws.Close() + return s.Write.Close() } func labelsToEqualityMatchers(ls model.LabelSet) []*labels.Matcher { diff --git a/vendor/github.com/prometheus/prometheus/storage/remote/write.go b/vendor/github.com/prometheus/prometheus/storage/remote/write.go index 78ab088ba785..11f6d358b167 100644 --- a/vendor/github.com/prometheus/prometheus/storage/remote/write.go +++ b/vendor/github.com/prometheus/prometheus/storage/remote/write.go @@ -60,8 +60,15 @@ type WriteStorage struct { samplesIn *ewmaRate flushDeadline time.Duration scraper scrape.ReadyManager + + // NewClient defaults to NewWriteClient when creating a WriteStorage. + NewClient WriteClientFunc } +// WriteClientFunc returns a WriteClient given the name of a remote write +// config. NewWriteClient implements the signature of this function. +type WriteClientFunc func(name string, conf *ClientConfig) (WriteClient, error) + // NewWriteStorage creates and runs a WriteStorage. func NewWriteStorage(logger log.Logger, reg prometheus.Registerer, walDir string, flushDeadline time.Duration, sm scrape.ReadyManager) *WriteStorage { if logger == nil { @@ -77,6 +84,8 @@ func NewWriteStorage(logger log.Logger, reg prometheus.Registerer, walDir string samplesIn: newEWMARate(ewmaWeight, shardUpdateDuration), walDir: walDir, scraper: sm, + + NewClient: NewWriteClient, } go rws.run() return rws @@ -122,7 +131,7 @@ func (rws *WriteStorage) ApplyConfig(conf *config.Config) error { name = rwConf.Name } - c, err := NewWriteClient(name, &ClientConfig{ + c, err := rws.NewClient(name, &ClientConfig{ URL: rwConf.URL, Timeout: rwConf.RemoteTimeout, HTTPClientConfig: rwConf.HTTPClientConfig, diff --git a/vendor/modules.txt b/vendor/modules.txt index f10279f533ca..2958515b44ac 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -525,7 +525,7 @@ github.com/prometheus/procfs/internal/util github.com/prometheus/procfs/nfs github.com/prometheus/procfs/sysfs github.com/prometheus/procfs/xfs -# github.com/prometheus/prometheus v1.8.2-0.20201105135750-00f16d1ac3a4 => github.com/grafana/prometheus v1.8.2-0.20201021200247-cf00050ed1e9 +# github.com/prometheus/prometheus v1.8.2-0.20201105135750-00f16d1ac3a4 => github.com/grafana/prometheus v1.8.2-0.20210111220521-c0d5de2f0ee3 github.com/prometheus/prometheus/config github.com/prometheus/prometheus/discovery github.com/prometheus/prometheus/discovery/azure