telegraf/plugins/inputs/http_listener/http_listener_test.go

package http_listener

import (
	"bytes"
	"crypto/tls"
	"crypto/x509"
	"io"
	"io/ioutil"
	"net/http"
	"net/url"
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/influxdata/telegraf/testutil"

	"github.com/stretchr/testify/require"
)

const (
	testMsg = "cpu_load_short,host=server01 value=12.0 1422568543702900257\n"

	testMsgNoNewline = "cpu_load_short,host=server01 value=12.0 1422568543702900257"

	testMsgs = `cpu_load_short,host=server02 value=12.0 1422568543702900257
cpu_load_short,host=server03 value=12.0 1422568543702900257
cpu_load_short,host=server04 value=12.0 1422568543702900257
cpu_load_short,host=server05 value=12.0 1422568543702900257
cpu_load_short,host=server06 value=12.0 1422568543702900257
`
	badMsg = "blahblahblah: 42\n"

	emptyMsg = ""

	serviceRootPEM = `-----BEGIN CERTIFICATE-----
MIIDRTCCAi2gAwIBAgIUenakcvMDj2URxBvUHBe0Mfhac0cwDQYJKoZIhvcNAQEL
BQAwGzEZMBcGA1UEAxMQdGVsZWdyYWYtdGVzdC1jYTAeFw0xNzA4MzEwNTE5NDNa
Fw0yNzA4MjkwNTIwMTNaMBsxGTAXBgNVBAMTEHRlbGVncmFmLXRlc3QtY2EwggEi
MA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDxpDlUEC6LNXQMhvTtlWKUekwa
xh2OaiR16WvO8iA+sYmjlpFXOe+V6YWT+daOGujCqlGdrfDjj3C3pqFPJ6Q4VXaA
xQyd0Ena7kRtuQ/IUSpTWxyrpSIzKL3dAoV0NYpjFWznjVMP3Rq4l+4cHqviZSvK
bWUK5n0vBGpEw3A22V9urhlSNkSbECvzn9EFHyIeJX603zaKXYw5wiDwCp1swbXW
2WS2h45JeI5xrpKcFmLaqRNe0swi6bkGnmefyCv7nsbOLeKyEW9AExDSd6nSLdu9
TGzhAfnfodcajSmKiQ+7YL9JY1bQ9hlfXk1ULg4riSEMKF+trZFZUanaXeeBAgMB
AAGjgYAwfjAOBgNVHQ8BAf8EBAMCAQYwDwYDVR0TAQH/BAUwAwEB/zAdBgNVHQ4E
FgQUiPkCD8gEsSgIiV8jzACMoUZcHaIwHwYDVR0jBBgwFoAUiPkCD8gEsSgIiV8j
zACMoUZcHaIwGwYDVR0RBBQwEoIQdGVsZWdyYWYtdGVzdC1jYTANBgkqhkiG9w0B
AQsFAAOCAQEAXeadR7ZVkb2C0F8OEd2CQxVt2/JOqM4G2N2O8uTwf+hIn+qm+jbb
Q6JokGhr5Ybhvtv3U9JnI6RVI+TOYNkDzs5e2DtntFQmcKb2c+y5Z+OpvWd13ObK
GMCs4bho6O7h1qo1Z+Ftd6sYQ7JL0MuTGWCNbXv2c1iC4zPT54n1vGZC5so08RO0
r7bqLLEnkSawabvSAeTxtweCXJUw3D576e0sb8oU0AP/Hn/2IC9E1vFZdjDswEfs
ARE4Oc5XnN6sqjtp0q5CqPpW6tYFwxdtZFk0VYPXyRnETVgry7Dc/iX6mktIYUx+
qWSyPEDKALyxx6yUyVDqgcY2VUm0rM/1Iw==
-----END CERTIFICATE-----`
	serviceCertPEM = `-----BEGIN CERTIFICATE-----
MIIDKjCCAhKgAwIBAgIUVYjQKruuFavlMZvV7X6RRF4OyBowDQYJKoZIhvcNAQEL
BQAwGzEZMBcGA1UEAxMQdGVsZWdyYWYtdGVzdC1jYTAeFw0xNzA4MzEwNTM3MjRa
Fw0xNzA5MzAwNTM3NTRaMBQxEjAQBgNVBAMTCWxvY2FsaG9zdDCCASIwDQYJKoZI
hvcNAQEBBQADggEPADCCAQoCggEBANojLHm+4ttLfl8xo4orZ436/o36wdQ30sWz
xE8eGejhARvCSNIR1Tau41Towq/MQVQQejQJRgqBSz7UEfzJNJGKKKc560j6fmTM
FHpFNZcTrNrTb0r3blUWF1oswhTgg313OXbVsz+E9tHkT1p/s9uURy3TJ3O/CFHq
2vTiTQMTq31v0FEN1E/d6uzMhnGy5+QuRu/0A2iPpgXgPopYZwG5t4hN1KklM//l
j2gMlX6mAYalctFOkDbhIe4/4dQcfT0sWA49KInZmUeB1RdyiNfCoXnDRZHocPIj
ltYAK/Igda0fdlMisoqh2ZMrCt8yhws7ycc12cFi7ZMv8zvi5p8CAwEAAaNtMGsw
EwYDVR0lBAwwCgYIKwYBBQUHAwEwHQYDVR0OBBYEFCdE87Nz7vPpgRmj++6J8rQR
0F/TMB8GA1UdIwQYMBaAFIj5Ag/IBLEoCIlfI8wAjKFGXB2iMBQGA1UdEQQNMAuC
CWxvY2FsaG9zdDANBgkqhkiG9w0BAQsFAAOCAQEAIPhMYCsCPvOcvLLkahaZVn2g
ZbzPDplFhEsH1cpc7vd3GCV2EYjNTbBTDs5NlovSbJLf1DFB+gwsfEjhlFVZB3UQ
6GtuA5CQh/Skv8ngCDiLP50BbKF0CLa4Ia0xrSTAyRsg2rt9APphbej0yKqJ7j8U
1KK6rjOSnuzrKseex26VVovjPFq0FgkghWRm0xrAeizGTBCSEStZEPhk3pBo2x95
a32VPpmhlQMDyiV6m1cc9/MfxMisnyeLqJl8E9nziNa4/BgwwN9DcOp63D9OOa6A
brtLz8OXqvV+7gKlq+nASFDimXwFKRyqRH6ECyHNTE2K14KZb7+JTa0AUm6Nlw==
-----END CERTIFICATE-----`
	serviceKeyPEM = `-----BEGIN RSA PRIVATE KEY-----
MIIEowIBAAKCAQEA2iMseb7i20t+XzGjiitnjfr+jfrB1DfSxbPETx4Z6OEBG8JI
0hHVNq7jVOjCr8xBVBB6NAlGCoFLPtQR/Mk0kYoopznrSPp+ZMwUekU1lxOs2tNv
SvduVRYXWizCFOCDfXc5dtWzP4T20eRPWn+z25RHLdMnc78IUera9OJNAxOrfW/Q
UQ3UT93q7MyGcbLn5C5G7/QDaI+mBeA+ilhnAbm3iE3UqSUz/+WPaAyVfqYBhqVy
0U6QNuEh7j/h1Bx9PSxYDj0oidmZR4HVF3KI18KhecNFkehw8iOW1gAr8iB1rR92
UyKyiqHZkysK3zKHCzvJxzXZwWLtky/zO+LmnwIDAQABAoIBABD8MidcrK9kpndl
FxXYIV0V0SJfBx6uJhRM1hlO/7d5ZauyqhbpWo/CeGMRKK+lmOShz9Ijcre4r5I5
0xi61gQLHPVAdkidcKAKoAGRSAX2ezwiwIS21Xl8md7ko0wa20I2uVu+chGdGdbo
DyG91dRgLFauHWFO26f9QIVW5aY6ifyjg1fyxR/9n2YZfkqbjvASW4Mmfv5GR1aT
mffajgsquy78PKs86f879iG+cfCzPYdoK+h7fsm4EEqDwK8JCsUIY1qN+Tuj5RQY
zuIuD34+wywe7Jd1vwjQ40Cyilgtnu8Q8s8J05bXrD3mqer5nrqIGOX0vKgs+EXx
1hV+6ZECgYEA+950L2u8oPzNXu9BAL8Y5Tl384qj1+Cj/g28MuZFoPf/KU0HRN6l
PBlXKaGP9iX+749tdiNPk5keIwOL8xCVXOpMLOA/jOlGODydG9rX67WCL/R1RcJR
+Pip8dxO1ZNpOKHud06XLMuuVz9qNq0Xwb1VCzNTOxnEDwtXNyDm6OkCgYEA3bcW
hMeDNn85UA4n0ljcdCmEu12WS7L//jaAOWuPPfM3GgKEIic6hqrPWEuZlZtQnybx
L6qQgaWyCfl/8z0+5WynQqkVPz1j1dSrSKnemxyeySrmUcOH5UJfAKaG5PUd7H3t
oPTCxkbW3Bi2QLlgd4nb7+OEk6w0V9Zzv4AFHkcCgYBL/aD2Ub4WoE9iLjNhg0aC
mmUrcI/gaSFxXDmE7d7iIxC0KE5iI/6cdFTM9bbWoD4bjx2KgDrZIGBsVfyaeE1o
PDSBcaMa46LRAtCv/8YXkqrVxx6+zlMnF/dGRp7uZ0xeztSA4JBR7p4KKtLj7jN1
u6b1+yVIdoylsVk+A8pHSQKBgQCUcsn5DTyleHl/SHsRM74naMUeToMbHDaalxMz
XvkBmZ8DIzwlQe7FzAgYLkYfDWblqMVEDQfERpT2aL9qtU8vfZhf4aYAObJmsYYd
mN8bLAaE2txrUmfi8JV7cgRPuG7YsVgxtK/U4glqRIGCxJv6bat86vERjvNc/JFz
XtwOcQKBgF83Ov+EA9pL0AFs+BMiO+0SX/OqLX0TDMSqUKg3jjVfgl+BSBEZIsOu
g5jqHBx3Om/UyrXdn+lnMhyEgCuNkeC6057B5iGcWucTlnINeejXk/pnbvMtGjD1
OGWmdXhgLtKg6Edqm+9fnH0UJN6DRxRRCUfzMfbY8TRmLzZG2W34
-----END RSA PRIVATE KEY-----`
	clientRootPEM = `-----BEGIN CERTIFICATE-----
MIIDRTCCAi2gAwIBAgIUenakcvMDj2URxBvUHBe0Mfhac0cwDQYJKoZIhvcNAQEL
BQAwGzEZMBcGA1UEAxMQdGVsZWdyYWYtdGVzdC1jYTAeFw0xNzA4MzEwNTE5NDNa
Fw0yNzA4MjkwNTIwMTNaMBsxGTAXBgNVBAMTEHRlbGVncmFmLXRlc3QtY2EwggEi
MA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDxpDlUEC6LNXQMhvTtlWKUekwa
xh2OaiR16WvO8iA+sYmjlpFXOe+V6YWT+daOGujCqlGdrfDjj3C3pqFPJ6Q4VXaA
xQyd0Ena7kRtuQ/IUSpTWxyrpSIzKL3dAoV0NYpjFWznjVMP3Rq4l+4cHqviZSvK
bWUK5n0vBGpEw3A22V9urhlSNkSbECvzn9EFHyIeJX603zaKXYw5wiDwCp1swbXW
2WS2h45JeI5xrpKcFmLaqRNe0swi6bkGnmefyCv7nsbOLeKyEW9AExDSd6nSLdu9
TGzhAfnfodcajSmKiQ+7YL9JY1bQ9hlfXk1ULg4riSEMKF+trZFZUanaXeeBAgMB
AAGjgYAwfjAOBgNVHQ8BAf8EBAMCAQYwDwYDVR0TAQH/BAUwAwEB/zAdBgNVHQ4E
FgQUiPkCD8gEsSgIiV8jzACMoUZcHaIwHwYDVR0jBBgwFoAUiPkCD8gEsSgIiV8j
zACMoUZcHaIwGwYDVR0RBBQwEoIQdGVsZWdyYWYtdGVzdC1jYTANBgkqhkiG9w0B
AQsFAAOCAQEAXeadR7ZVkb2C0F8OEd2CQxVt2/JOqM4G2N2O8uTwf+hIn+qm+jbb
Q6JokGhr5Ybhvtv3U9JnI6RVI+TOYNkDzs5e2DtntFQmcKb2c+y5Z+OpvWd13ObK
GMCs4bho6O7h1qo1Z+Ftd6sYQ7JL0MuTGWCNbXv2c1iC4zPT54n1vGZC5so08RO0
r7bqLLEnkSawabvSAeTxtweCXJUw3D576e0sb8oU0AP/Hn/2IC9E1vFZdjDswEfs
ARE4Oc5XnN6sqjtp0q5CqPpW6tYFwxdtZFk0VYPXyRnETVgry7Dc/iX6mktIYUx+
qWSyPEDKALyxx6yUyVDqgcY2VUm0rM/1Iw==
-----END CERTIFICATE-----`
	clientCertPEM = `-----BEGIN CERTIFICATE-----
MIIDMDCCAhigAwIBAgIUIVOF5g2zH6+J/dbGdu4q18aSJoMwDQYJKoZIhvcNAQEL
BQAwGzEZMBcGA1UEAxMQdGVsZWdyYWYtdGVzdC1jYTAeFw0xNzA4MzEwNTQ1MzJa
Fw0yNzA4MjUwMTQ2MDJaMBcxFTATBgNVBAMTDGR1bW15LWNsaWVudDCCASIwDQYJ
KoZIhvcNAQEBBQADggEPADCCAQoCggEBAKok1HJ40buyjrS+DG9ORLzrWIJad2y/
6X2Bg9MSENfpEUgaS7nK2ML3m1e2poHqBSR+V8VECNs+MDCLSOeQ4FC1TdBKMLfw
NxW88y5Gj6rTRcAXl092ba7stwbqJPBAZu1Eh1jXIp5nrFKh8Jq7kRxmMB5vC70V
fOSPS0RZtEd7D+QZ6jgkFJWsZzn4gJr8nc/kmLcntLw+g/tz9/8lfaV306tLlhMH
dv3Ka6Nt86j6/muOwvoeAkAnCEFAgDcXg4F37PFAiEHRw9DyTeWDuZqvnMZ3gosL
kl15QhnP0yG2QCjSb1gaLcKB42cyxDnPc31WsVuuzQnajazcVf3lJW0CAwEAAaNw
MG4wEwYDVR0lBAwwCgYIKwYBBQUHAwIwHQYDVR0OBBYEFCemMO+Qlj+YCLQ3ScAQ
8XYJJJ5ZMB8GA1UdIwQYMBaAFIj5Ag/IBLEoCIlfI8wAjKFGXB2iMBcGA1UdEQQQ
MA6CDGR1bW15LWNsaWVudDANBgkqhkiG9w0BAQsFAAOCAQEARThbApKvvGDp7uSc
mINaqDOHe69F9PepV0/3+B5+X1b3yd2sbzZL/ZoHl27kajSHVrUF+09gcTosfuY3
omnIPw+NseqTJG+qTMRb3AarLNO46EJZLOowAEhnJyVmhK5uU0YqhV1X9eN+g4/o
BuyOPvHj6UJWviZFy6fDIj2N+ygN/CNP5X3iLDBUoyCEHAehLiQr0aRgsqe4JLlS
P+0l0btTUpcqUhsQy+sD2lv3MO1tZ/P4zhzu0J0LUeLBDdOPf/FIvTgkCNxN9GGy
SLmeBeCzsKmWbzE3Yuahw3h4IblVyyGc7ZDGIobDrZgFqshcZylU8wrsjUnjNSPA
G+LOWQ==
-----END CERTIFICATE-----`
	clientKeyPEM = `-----BEGIN RSA PRIVATE KEY-----
MIIEogIBAAKCAQEAqiTUcnjRu7KOtL4Mb05EvOtYglp3bL/pfYGD0xIQ1+kRSBpL
ucrYwvebV7amgeoFJH5XxUQI2z4wMItI55DgULVN0Eowt/A3FbzzLkaPqtNFwBeX
T3Ztruy3Buok8EBm7USHWNcinmesUqHwmruRHGYwHm8LvRV85I9LRFm0R3sP5Bnq
OCQUlaxnOfiAmvydz+SYtye0vD6D+3P3/yV9pXfTq0uWEwd2/cpro23zqPr+a47C
+h4CQCcIQUCANxeDgXfs8UCIQdHD0PJN5YO5mq+cxneCiwuSXXlCGc/TIbZAKNJv
WBotwoHjZzLEOc9zfVaxW67NCdqNrNxV/eUlbQIDAQABAoIBAAXZYEhTKPqn58oE
4o6NBUXtXUyV6ZcefdtnsW13KIcTpxlwdfv8IjmJo5h/WfgLYIPhqAjLDvbii2uP
zkDPtTZxFSy88DHSm0IvDbkgid3Yh4RUC0qbCqhB0QT21bBAtokfmvuN4c3KSJ1K
nefj3Ng6Fxtku+WTMIj2+CJwZwcyAH47ZUngYs/77gA0hAJcbdL/bj8Bpmd+lH6C
Ci22T2hrw+cpWMN6qwa3wxWIneCaqxkylSgpUzSNE0QO3mXkX+NYtL2BQ0w+wPqq
lww3QJOFAX1qRLflglL9K+ruTQofm49vxv6apsoqdkrxEBoPzkljlqiPRmzUxau4
cvbApQECgYEAy5m5O3mQt6DBrDRJWSwpZh6DRNd5USEqXOIFtp+Qze2Jx1pYQfZt
NOXOrwy04o0+6yLzc4O4W5ta2KfTlALFzCa6Na3Ca4ZUAeteWprrdh8b1b2w/wUH
E3uQFkvH0zFdPsA3pTTZ0k/ydmHnu4zZqBnSeh0dIW8xFYgZZCgQusECgYEA1e7O
ujCUa8y49sY42D/Y/c8B96xVfJZO5hhY7eLgkzqUlmFl31Ld7AjlJcXpbMeW1vaa
0Mxbfx2qAVaZEkvdnXq3V8spe6qOGBdlKzey4DMEfmEXLFp5DRYCSwpXiqDZcGqc
jwI58wuzKoDgydN9bLdF8XYGtQXnHIE9WyTYMa0CgYBKYSBgb+rEir/2LyvUneOJ
4P/HuIgjcWBOimvX6bc2495/q6uufV4sAwBcxuGWGk+wCxaxTp+dJ8YqfDU5T0H/
cO56Cb6LFYm/IcNYilwWzQqYLTJqF+Yb4fojiw+3QcN01zf87K/eu0IyqVXFGJGz
bauM3PH1cu+VlCDijBiAgQKBgDOQ9YmRriTx2t+41fjiIvbC0BGYG58FSA1UbxMg
LcuvQiOhZIHZIp8DYeCh/Or4jRZRqO2NZLyWNOVPr2Pmn4uXCdyCnwQtD0UlVoB9
U4ORKJMh6gkJ4cXSuUjHPGSw8tiTChu6iKdZ+ZzUJdrgPIpY/uX98g3uV0/aoyR2
FBqdAoGAQIrcOsTpCe6l3ZDtQyNIeAj1s7sZyW8MBy95RXw3y/yzVEOAu4yWNobj
RReeHQEsrQq+sJ/cols8HfoOpGpL3U0IGDi5vr1JlOXmBhFX2xuFrfh3jvgXlUqb
fqxPcT3d7I/UEi0ueDh3osyTn46mDfRfF7HBLBNeyQbIFWBDDus=
-----END RSA PRIVATE KEY-----`
)

var (
	initClient           sync.Once
	client               *http.Client
	initServiceCertFiles sync.Once
	allowedCAFiles       []string
	serviceCAFiles       []string
	serviceCertFile      string
	serviceKeyFile       string
)

func newTestHTTPListener() *HTTPListener {
	listener := &HTTPListener{
		ServiceAddress: ":0",
	}
	return listener
}

func newTestHTTPSListener() *HTTPListener {
	initServiceCertFiles.Do(func() {
		acaf, err := ioutil.TempFile("", "allowedCAFile.crt")
		if err != nil {
			panic(err)
		}
		defer acaf.Close()
		_, err = io.Copy(acaf, bytes.NewReader([]byte(clientRootPEM)))
		allowedCAFiles = []string{acaf.Name()}

		scaf, err := ioutil.TempFile("", "serviceCAFile.crt")
		if err != nil {
			panic(err)
		}
		defer scaf.Close()
		_, err = io.Copy(scaf, bytes.NewReader([]byte(serviceRootPEM)))
		serviceCAFiles = []string{scaf.Name()}

		scf, err := ioutil.TempFile("", "serviceCertFile.crt")
		if err != nil {
			panic(err)
		}
		defer scf.Close()
		_, err = io.Copy(scf, bytes.NewReader([]byte(serviceCertPEM)))
		serviceCertFile = scf.Name()

		skf, err := ioutil.TempFile("", "serviceKeyFile.crt")
		if err != nil {
			panic(err)
		}
		defer skf.Close()
		_, err = io.Copy(skf, bytes.NewReader([]byte(serviceKeyPEM)))
		serviceKeyFile = skf.Name()
	})

	listener := &HTTPListener{
		ServiceAddress:    ":0",
		TlsAllowedCacerts: allowedCAFiles,
		TlsCert:           serviceCertFile,
		TlsKey:            serviceKeyFile,
	}

	return listener
}

func getHTTPSClient() *http.Client {
	initClient.Do(func() {
		cas := x509.NewCertPool()
		cas.AppendCertsFromPEM([]byte(serviceRootPEM))
		clientCert, err := tls.X509KeyPair([]byte(clientCertPEM), []byte(clientKeyPEM))
		if err != nil {
			panic(err)
		}
		client = &http.Client{
			Transport: &http.Transport{
				TLSClientConfig: &tls.Config{
					RootCAs:            cas,
					Certificates:       []tls.Certificate{clientCert},
					MinVersion:         tls.VersionTLS12,
					MaxVersion:         tls.VersionTLS12,
					CipherSuites:       []uint16{tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256},
					Renegotiation:      tls.RenegotiateNever,
					InsecureSkipVerify: false,
				},
			},
		}
	})
	return client
}

func createURL(listener *HTTPListener, scheme string, path string, rawquery string) string {
	u := url.URL{
		Scheme:   scheme,
		Host:     "localhost:" + strconv.Itoa(listener.Port),
		Path:     path,
		RawQuery: rawquery,
	}
	return u.String()
}

func TestWriteHTTPSNoClientAuth(t *testing.T) {
	listener := newTestHTTPSListener()
	listener.TlsAllowedCacerts = nil

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	cas := x509.NewCertPool()
	cas.AppendCertsFromPEM([]byte(serviceRootPEM))
	noClientAuthClient := &http.Client{
		Transport: &http.Transport{
			TLSClientConfig: &tls.Config{
				RootCAs: cas,
			},
		},
	}

	// post single message to listener
	resp, err := noClientAuthClient.Post(createURL(listener, "https", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)
}

func TestWriteHTTPSWithClientAuth(t *testing.T) {
	listener := newTestHTTPSListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post single message to listener
	resp, err := getHTTPSClient().Post(createURL(listener, "https", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)
}

func TestWriteHTTP(t *testing.T) {
	listener := newTestHTTPListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post single message to listener
	resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)

	acc.Wait(1)
	acc.AssertContainsTaggedFields(t, "cpu_load_short",
		map[string]interface{}{"value": float64(12)},
		map[string]string{"host": "server01"},
	)

	// post multiple message to listener
	resp, err = http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsgs)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)

	acc.Wait(2)
	hostTags := []string{"server02", "server03",
		"server04", "server05", "server06"}
	for _, hostTag := range hostTags {
		acc.AssertContainsTaggedFields(t, "cpu_load_short",
			map[string]interface{}{"value": float64(12)},
			map[string]string{"host": hostTag},
		)
	}

	// Post a gigantic metric to the listener and verify that an error is returned:
	resp, err = http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(hugeMetric)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 400, resp.StatusCode)

	acc.Wait(3)
	acc.AssertContainsTaggedFields(t, "cpu_load_short",
		map[string]interface{}{"value": float64(12)},
		map[string]string{"host": "server01"},
	)
}

// http listener should add a newline at the end of the buffer if it's not there
func TestWriteHTTPNoNewline(t *testing.T) {
	listener := newTestHTTPListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post single message to listener
	resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsgNoNewline)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)

	acc.Wait(1)
	acc.AssertContainsTaggedFields(t, "cpu_load_short",
		map[string]interface{}{"value": float64(12)},
		map[string]string{"host": "server01"},
	)
}

func TestWriteHTTPMaxLineSizeIncrease(t *testing.T) {
	listener := &HTTPListener{
		ServiceAddress: ":0",
		MaxLineSize:    128 * 1000,
	}

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// Post a gigantic metric to the listener and verify that it writes OK this time:
	resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(hugeMetric)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)
}

func TestWriteHTTPVerySmallMaxBody(t *testing.T) {
	listener := &HTTPListener{
		ServiceAddress: ":0",
		MaxBodySize:    4096,
	}

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	resp, err := http.Post(createURL(listener, "http", "/write", ""), "", bytes.NewBuffer([]byte(hugeMetric)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 413, resp.StatusCode)
}

func TestWriteHTTPVerySmallMaxLineSize(t *testing.T) {
	listener := &HTTPListener{
		ServiceAddress: ":0",
		MaxLineSize:    70,
	}

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	resp, err := http.Post(createURL(listener, "http", "/write", ""), "", bytes.NewBuffer([]byte(testMsgs)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)

	hostTags := []string{"server02", "server03",
		"server04", "server05", "server06"}
	acc.Wait(len(hostTags))
	for _, hostTag := range hostTags {
		acc.AssertContainsTaggedFields(t, "cpu_load_short",
			map[string]interface{}{"value": float64(12)},
			map[string]string{"host": hostTag},
		)
	}
}

func TestWriteHTTPLargeLinesSkipped(t *testing.T) {
	listener := &HTTPListener{
		ServiceAddress: ":0",
		MaxLineSize:    100,
	}

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	resp, err := http.Post(createURL(listener, "http", "/write", ""), "", bytes.NewBuffer([]byte(hugeMetric+testMsgs)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 400, resp.StatusCode)

	hostTags := []string{"server02", "server03",
		"server04", "server05", "server06"}
	acc.Wait(len(hostTags))
	for _, hostTag := range hostTags {
		acc.AssertContainsTaggedFields(t, "cpu_load_short",
			map[string]interface{}{"value": float64(12)},
			map[string]string{"host": hostTag},
		)
	}
}

// test that writing gzipped data works
func TestWriteHTTPGzippedData(t *testing.T) {
	listener := newTestHTTPListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	data, err := ioutil.ReadFile("./testdata/testmsgs.gz")
	require.NoError(t, err)

	req, err := http.NewRequest("POST", createURL(listener, "http", "/write", ""), bytes.NewBuffer(data))
	require.NoError(t, err)
	req.Header.Set("Content-Encoding", "gzip")

	client := &http.Client{}
	resp, err := client.Do(req)
	require.NoError(t, err)
	require.EqualValues(t, 204, resp.StatusCode)

	hostTags := []string{"server02", "server03",
		"server04", "server05", "server06"}
	acc.Wait(len(hostTags))
	for _, hostTag := range hostTags {
		acc.AssertContainsTaggedFields(t, "cpu_load_short",
			map[string]interface{}{"value": float64(12)},
			map[string]string{"host": hostTag},
		)
	}
}

// writes 25,000 metrics to the listener with 10 different writers
func TestWriteHTTPHighTraffic(t *testing.T) {
	listener := newTestHTTPListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post many messages to listener
	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(innerwg *sync.WaitGroup) {
			defer innerwg.Done()
			for i := 0; i < 500; i++ {
				resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsgs)))
				require.NoError(t, err)
				resp.Body.Close()
				require.EqualValues(t, 204, resp.StatusCode)
			}
		}(&wg)
	}

	wg.Wait()
	listener.Gather(acc)

	acc.Wait(25000)
	require.Equal(t, int64(25000), int64(acc.NMetrics()))
}

func TestReceive404ForInvalidEndpoint(t *testing.T) {
	listener := newTestHTTPListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post single message to listener
	resp, err := http.Post(createURL(listener, "http", "/foobar", ""), "", bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 404, resp.StatusCode)
}

func TestWriteHTTPInvalid(t *testing.T) {
	listener := newTestHTTPListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post single message to listener
	resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(badMsg)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 400, resp.StatusCode)
}

func TestWriteHTTPEmpty(t *testing.T) {
	listener := newTestHTTPListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post single message to listener
	resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(emptyMsg)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)
}

func TestQueryAndPingHTTP(t *testing.T) {
	listener := newTestHTTPListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post query to listener
	resp, err := http.Post(
		createURL(listener, "http", "/query", "db=&q=CREATE+DATABASE+IF+NOT+EXISTS+%22mydb%22"), "", nil)
	require.NoError(t, err)
	require.EqualValues(t, 200, resp.StatusCode)

	// post ping to listener
	resp, err = http.Post(createURL(listener, "http", "/ping", ""), "", nil)
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)
}

func TestWriteWithPrecision(t *testing.T) {
	listener := newTestHTTPListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	msg := "xyzzy value=42 1422568543\n"
	resp, err := http.Post(
		createURL(listener, "http", "/write", "precision=s"), "", bytes.NewBuffer([]byte(msg)))
	require.NoError(t, err)
	resp.Body.Close()
	require.EqualValues(t, 204, resp.StatusCode)

	acc.Wait(1)
	require.Equal(t, 1, len(acc.Metrics))
	require.Equal(t, time.Unix(0, 1422568543000000000), acc.Metrics[0].Time)
}

const hugeMetric = `super_long_metric,foo=bar clients=1i,connected_slaves=0i,evicted_keys=0i,expired_keys=0i,instantaneous_ops_per_sec=0i,keyspace_hitrate=0,keyspace_hits=0i,keyspace_misses=2i,latest_fork_usec=0i,master_repl_offset=0i,mem_fragmentation_ratio=3.58,pubsub_channels=0i,pubsub_patterns=0i,rdb_changes_since_last_save=0i,repl_backlog_active=0i,repl_backlog_histlen=0i,repl_backlog_size=1048576i,sync_full=0i,sync_partial_err=0i,sync_partial_ok=0i,total_commands_processed=4i,total_connections_received=2i,uptime=869i,used_cpu_sys=0.07,used_cpu_sys_children=0,used_cpu_user=0.1,used_cpu_user_children=0,used_memory=502048i,used_memory_lua=33792i,used_memory_peak=501128i,used_memory_rss=1798144i,clients=1i,connected_slaves=0i,evicted_keys=0i,expired_keys=0i,instantaneous_ops_per_sec=0i,keyspace_hitrate=0,keyspace_hits=0i,keyspace_misses=2i,latest_fork_usec=0i,master_repl_offset=0i,mem_fragmentation_ratio=3.58,pubsub_channels=0i,pubsub_patterns=0i,rdb_changes_since_last_save=0i,repl_backlog_active=0i,repl_backlog_histlen=0i,repl_backlog_size=1048576i,sync_full=0i,sync_partial_err=0i,sync_partial_ok=0i,total_commands_processed=4i,total_connections_received=2i,uptime=869i,used_cpu_sys=0.07,used_cpu_sys_children=0,used_cpu_user=0.1,used_cpu_user_children=0,used_memory=502048i,used_memory_lua=33792i,used_memory_peak=501128i,used_memory_rss=1798144i,clients=1i,connected_slaves=0i,evicted_keys=0i,expired_keys=0i,instantaneous_ops_per_sec=0i,keyspace_hitrate=0,keyspace_hits=0i,keyspace_misses=2i,latest_fork_usec=0i,master_repl_offset=0i,mem_fragmentation_ratio=3.58,pubsub_channels=0i,pubsub_patterns=0i,rdb_changes_since_last_save=0i,repl_backlog_active=0i,repl_backlog_histlen=0i,repl_backlog_size=1048576i,sync_full=0i,sync_partial_err=0i,sync_partial_ok=0i,total_commands_processed=4i,total_connections_received=2i,uptime=869i,used_cpu_sys=0.07,used_cpu_sys_children=0,used_cpu_user=0.1,used_cpu_user_children=0,used_memory=502048i,used_memory_lua=33792i,used_memory_peak=501128i,used_memory_rss=1798144i,clients=1i,connected_slaves=0i,evicted_keys=0i,expired_keys=0i,instantaneous_ops_per_sec=0i,keyspace_hitrate=0,keyspace_hits=0i,keyspace_misses=2i,latest_fork_usec=0i,master_repl_offset=0i,mem_fragmentation_ratio=3.58,pubsub_channels=0i,pubsub_patterns=0i,rdb_changes_since_last_save=0i,repl_backlog_active=0i,repl_backlog_histlen=0i,repl_backlog_size=1048576i,sync_full=0i,sync_partial_err=0i,sync_partial_ok=0i,total_commands_processed=4i,total_connections_received=2i,uptime=869i,used_cpu_sys=0.07,used_cpu_sys_children=0,used_cpu_user=0.1,used_cpu_user_children=0,used_memory=502048i,used_memory_lua=33792i,used_memory_peak=501128i,used_memory_rss=1798144i,clients=1i,connected_slaves=0i,evicted_keys=0i,expired_keys=0i,instantaneous_ops_per_sec=0i,keyspace_hitrate=0,keyspace_hits=0i,keyspace_misses=2i,latest_fork_usec=0i,master_repl_offset=0i,mem_fragmentation_ratio=3.58,pubsub_channels=0i,pubsub_patterns=0i,rdb_changes_since_last_save=0i,repl_backlog_active=0i,repl_backlog_histlen=0i,repl_backlog_size=1048576i,sync_full=0i,sync_partial_err=0i,sync_partial_ok=0i,total_commands_processed=4i,total_connections_received=2i,uptime=869i,used_cpu_sys=0.07,used_cpu_sys_children=0,used_cpu_user=0.1,used_cpu_user_children=0,used_memory=502048i,used_memory_lua=33792i,used_memory_peak=501128i,used_memory_rss=1798144i,clients=1i,connected_slaves=0i,evicted_keys=0i,expired_keys=0i,instantaneous_ops_per_sec=0i,keyspace_hitrate=0,keyspace_hits=0i,keyspace_misses=2i,latest_fork_usec=0i,master_repl_offset=0i,mem_fragmentation_ratio=3.58,pubsub_channels=0i,pubsub_patterns=0i,rdb_changes_since_last_save=0i,repl_backlog_active=0i,repl_backlog_histlen=0i,repl_backlog_size=1048576i,sync_full=0i,sync_partial_err=0i,sync_partial_ok=0i,total_commands_processed=4i,total_connections_received=2i,uptime=869i,used_cpu_sys=0.07,used_cpu_sys_children=0,used_cpu_user=0.1,used_cpu_user_children=0,used_memory=502048i,used_memory_lua=33792i,used_memory_peak=501128i,used_memory_rss=1798144i,clients=1i,connected_slaves=0i,evicted_keys=0i,expired_keys=0i,instantaneous_ops_pe
`