Add ability to specify HTTP Headers in http_listener_v2 which will ad…

…ded as tags (influxdata#7223)

plugins/inputs/http_listener_v2/README.md

@@ -48,6 +48,11 @@ This is a sample configuration for the plugin.
   # basic_username = "foobar"
   # basic_password = "barfoo"
 
+  ## Optional setting to map http headers into tags
+  ## If the http header is not present on the request, no corresponding tag will be added
+  ## If multiple instances of the http header are present, only the first value will be used
+  # http_header_tags = {"HTTP_HEADER" = "TAG_NAME"}
+
   ## Data format to consume.
   ## Each data format has its own unique set of configuration options, read
   ## more about them here:

plugins/inputs/http_listener_v2/http_listener_v2.go

@@ -44,6 +44,7 @@ type HTTPListenerV2 struct {
 	Port           int               `toml:"port"`
 	BasicUsername  string            `toml:"basic_username"`
 	BasicPassword  string            `toml:"basic_password"`
+	HTTPHeaderTags map[string]string `toml:"http_header_tags"`
 	tlsint.ServerConfig
 
 	TimeFunc
@@ -93,6 +94,11 @@ const sampleConfig = `
   # basic_username = "foobar"
   # basic_password = "barfoo"
 
+  ## Optional setting to map http headers into tags
+  ## If the http header is not present on the request, no corresponding tag will be added
+  ## If multiple instances of the http header are present, only the first value will be used
+  # http_header_tags = {"HTTP_HEADER" = "TAG_NAME"}
+
   ## Data format to consume.
   ## Each data format has its own unique set of configuration options, read
   ## more about them here:
@@ -225,6 +231,13 @@ func (h *HTTPListenerV2) serveWrite(res http.ResponseWriter, req *http.Request)
 	}
 
 	for _, m := range metrics {
+		for headerName, measurementName := range h.HTTPHeaderTags {
+			headerValues, foundHeader := req.Header[headerName]
+			if foundHeader && len(headerValues) > 0 {
+				m.AddTag(measurementName, headerValues[0])
+			}
+		}
+
 		h.acc.AddMetric(m)
 	}
 

plugins/inputs/http_listener_v2/http_listener_v2_test.go

@@ -381,6 +381,73 @@ func TestWriteHTTPEmpty(t *testing.T) {
 	require.EqualValues(t, 204, resp.StatusCode)
 }
 
+func TestWriteHTTPTransformHeaderValuesToTagsSingleWrite(t *testing.T) {
+	listener := newTestHTTPListenerV2()
+	listener.HTTPHeaderTags = map[string]string{"Present_http_header_1": "presentMeasurementKey1", "Present_http_header_2": "presentMeasurementKey2", "NOT_PRESENT_HEADER": "notPresentMeasurementKey"}
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	req, err := http.NewRequest("POST", createURL(listener, "http", "/write", "db=mydb"), bytes.NewBuffer([]byte(testMsg)))
+	require.NoError(t, err)
+	req.Header.Set("Content-Type", "")
+	req.Header.Set("Present_http_header_1", "PRESENT_HTTP_VALUE_1")
+	req.Header.Set("Present_http_header_2", "PRESENT_HTTP_VALUE_2")
+
+	resp, err := http.DefaultClient.Do(req)
+	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", "presentMeasurementKey1": "PRESENT_HTTP_VALUE_1", "presentMeasurementKey2": "PRESENT_HTTP_VALUE_2"},
+	)
+
+	// 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", "presentMeasurementKey1": "PRESENT_HTTP_VALUE_1", "presentMeasurementKey2": "PRESENT_HTTP_VALUE_2"},
+	)
+}
+
+func TestWriteHTTPTransformHeaderValuesToTagsBulkWrite(t *testing.T) {
+	listener := newTestHTTPListenerV2()
+	listener.HTTPHeaderTags = map[string]string{"Present_http_header_1": "presentMeasurementKey1", "Present_http_header_2": "presentMeasurementKey2", "NOT_PRESENT_HEADER": "notPresentMeasurementKey"}
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	req, err := http.NewRequest("POST", createURL(listener, "http", "/write", "db=mydb"), bytes.NewBuffer([]byte(testMsgs)))
+	require.NoError(t, err)
+	req.Header.Set("Content-Type", "")
+	req.Header.Set("Present_http_header_1", "PRESENT_HTTP_VALUE_1")
+	req.Header.Set("Present_http_header_2", "PRESENT_HTTP_VALUE_2")
+
+	resp, err := http.DefaultClient.Do(req)
+	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, "presentMeasurementKey1": "PRESENT_HTTP_VALUE_1", "presentMeasurementKey2": "PRESENT_HTTP_VALUE_2"},
+		)
+	}
+}
+
 func TestWriteHTTPQueryParams(t *testing.T) {
 	parser, _ := parsers.NewFormUrlencodedParser("query_measurement", nil, []string{"tagKey"})
 	listener := newTestHTTPListenerV2()