Add OAuth2 support to HTTP output plugin (#4536)

plugins/outputs/http/README.md

@@ -21,6 +21,12 @@ data formats.  For data_formats that support batching, metrics are sent in batch
   # username = "username"
   # password = "pa$$word"
 
+  ## OAuth2 Client Credentials Grant
+  # client_id = "clientid"
+  # client_secret = "secret"
+  # token_url = "https://indentityprovider/oauth2/v1/token"
+  # scopes = ["urn:opc:idm:__myscopes__"]
+
   ## Optional TLS Config
   # tls_ca = "/etc/telegraf/ca.pem"
   # tls_cert = "/etc/telegraf/cert.pem"
@@ -33,7 +39,7 @@ data formats.  For data_formats that support batching, metrics are sent in batch
   ## more about them here:
   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
   # data_format = "influx"
-  
+
   ## Additional HTTP headers
   # [outputs.http.headers]
   #   # Should be set manually to "application/json" for json data_format

plugins/outputs/http/http.go

@@ -2,6 +2,7 @@ package http
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -13,6 +14,8 @@ import (
 	"github.com/influxdata/telegraf/internal/tls"
 	"github.com/influxdata/telegraf/plugins/outputs"
 	"github.com/influxdata/telegraf/plugins/serializers"
+	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/clientcredentials"
 )
 
 var sampleConfig = `
@@ -29,6 +32,12 @@ var sampleConfig = `
   # username = "username"
   # password = "pa$$word"
 
+  ## OAuth2 Client Credentials Grant
+  # client_id = "clientid"
+  # client_secret = "secret"
+  # token_url = "https://indentityprovider/oauth2/v1/token"
+  # scopes = ["urn:opc:idm:__myscopes__"]
+
   ## Optional TLS Config
   # tls_ca = "/etc/telegraf/ca.pem"
   # tls_cert = "/etc/telegraf/cert.pem"
@@ -41,7 +50,7 @@ var sampleConfig = `
   ## more about them here:
   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
   # data_format = "influx"
-  
+
   ## Additional HTTP headers
   # [outputs.http.headers]
   #   # Should be set manually to "application/json" for json data_format
@@ -55,12 +64,16 @@ const (
 )
 
 type HTTP struct {
-	URL      string            `toml:"url"`
-	Timeout  internal.Duration `toml:"timeout"`
-	Method   string            `toml:"method"`
-	Username string            `toml:"username"`
-	Password string            `toml:"password"`
-	Headers  map[string]string `toml:"headers"`
+	URL          string            `toml:"url"`
+	Timeout      internal.Duration `toml:"timeout"`
+	Method       string            `toml:"method"`
+	Username     string            `toml:"username"`
+	Password     string            `toml:"password"`
+	Headers      map[string]string `toml:"headers"`
+	ClientID     string            `toml:"client_id"`
+	ClientSecret string            `toml:"client_secret"`
+	TokenURL     string            `toml:"token_url"`
+	Scopes       []string          `toml:"scopes"`
 	tls.ClientConfig
 
 	client     *http.Client
@@ -71,6 +84,34 @@ func (h *HTTP) SetSerializer(serializer serializers.Serializer) {
 	h.serializer = serializer
 }
 
+func (h *HTTP) createClient(ctx context.Context) (*http.Client, error) {
+	tlsCfg, err := h.ClientConfig.TLSConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	client := &http.Client{
+		Transport: &http.Transport{
+			TLSClientConfig: tlsCfg,
+			Proxy:           http.ProxyFromEnvironment,
+		},
+		Timeout: h.Timeout.Duration,
+	}
+
+	if h.ClientID != "" && h.ClientSecret != "" && h.TokenURL != "" {
+		oauthConfig := clientcredentials.Config{
+			ClientID:     h.ClientID,
+			ClientSecret: h.ClientSecret,
+			TokenURL:     h.TokenURL,
+			Scopes:       h.Scopes,
+		}
+		ctx = context.WithValue(ctx, oauth2.HTTPClient, client)
+		client = oauthConfig.Client(ctx)
+	}
+
+	return client, nil
+}
+
 func (h *HTTP) Connect() error {
 	if h.Method == "" {
 		h.Method = http.MethodPost
@@ -84,18 +125,13 @@ func (h *HTTP) Connect() error {
 		h.Timeout.Duration = defaultClientTimeout
 	}
 
-	tlsCfg, err := h.ClientConfig.TLSConfig()
+	ctx := context.Background()
+	client, err := h.createClient(ctx)
 	if err != nil {
 		return err
 	}
 
-	h.client = &http.Client{
-		Transport: &http.Transport{
-			TLSClientConfig: tlsCfg,
-			Proxy:           http.ProxyFromEnvironment,
-		},
-		Timeout: h.Timeout.Duration,
-	}
+	h.client = client
 
 	return nil
 }

plugins/outputs/http/http_test.go

@@ -287,3 +287,76 @@ func TestBasicAuth(t *testing.T) {
 		})
 	}
 }
+
+type TestHandlerFunc func(t *testing.T, w http.ResponseWriter, r *http.Request)
+
+func TestOAuthClientCredentialsGrant(t *testing.T) {
+	ts := httptest.NewServer(http.NotFoundHandler())
+	defer ts.Close()
+
+	var token = "2YotnFZFEjr1zCsicMWpAA"
+
+	u, err := url.Parse(fmt.Sprintf("http://%s", ts.Listener.Addr().String()))
+	require.NoError(t, err)
+
+	tests := []struct {
+		name         string
+		plugin       *HTTP
+		tokenHandler TestHandlerFunc
+		handler      TestHandlerFunc
+	}{
+		{
+			name: "no credentials",
+			plugin: &HTTP{
+				URL: u.String(),
+			},
+			handler: func(t *testing.T, w http.ResponseWriter, r *http.Request) {
+				require.Len(t, r.Header["Authorization"], 0)
+				w.WriteHeader(http.StatusOK)
+			},
+		},
+		{
+			name: "success",
+			plugin: &HTTP{
+				URL:          u.String() + "/write",
+				ClientID:     "howdy",
+				ClientSecret: "secret",
+				TokenURL:     u.String() + "/token",
+				Scopes:       []string{"urn:opc:idm:__myscopes__"},
+			},
+			tokenHandler: func(t *testing.T, w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				values := url.Values{}
+				values.Add("access_token", token)
+				values.Add("token_type", "bearer")
+				values.Add("expires_in", "3600")
+				w.Write([]byte(values.Encode()))
+			},
+			handler: func(t *testing.T, w http.ResponseWriter, r *http.Request) {
+				require.Equal(t, []string{"Bearer " + token}, r.Header["Authorization"])
+				w.WriteHeader(http.StatusOK)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ts.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				switch r.URL.Path {
+				case "/write":
+					tt.handler(t, w, r)
+				case "/token":
+					tt.tokenHandler(t, w, r)
+				}
+			})
+
+			serializer := influx.NewSerializer()
+			tt.plugin.SetSerializer(serializer)
+			err = tt.plugin.Connect()
+			require.NoError(t, err)
+
+			err = tt.plugin.Write([]telegraf.Metric{getMetric()})
+			require.NoError(t, err)
+		})
+	}
+}