Add support for HTTP basic auth (#24)

* Add support for HTTP basic auth

README.md

@@ -15,6 +15,8 @@ Configure the provider directly, or set the ENV variable `KAFKA_CONNECT_URL`
 ```hcl
 provider "kafka-connect" {
   url = "http://localhost:8083"
+  basic_auth_username = "user" # Optional
+  basic_auth_password = "password" # Optional
 }
 
 resource "kafka-connect_connector" "sqlite-sink" {
@@ -36,6 +38,22 @@ resource "kafka-connect_connector" "sqlite-sink" {
 }
 ```
 
+## Provider Properties
+
+| Property              | Type   | Example                 | Alternative environment variable name |
+|-----------------------|--------|-------------------------|---------------------------------------|
+| `url`                 | URL    | "http://localhost:8083" | `KAFKA_CONNECT_URL`                   |
+| `basic_auth_username` | String | "user"                  | `KAFKA_CONNECT_BASIC_AUTH_USERNAME`   |
+| `basic_auth_password` | String | "password"              | `KAFKA_CONNECT_BASIC_AUTH_PASSWORD`   |
+
+## Resource Properties
+
+| Property              | Type      | Description                                                          |
+|-----------------------|-----------|----------------------------------------------------------------------|
+| `name`                | String    | Connector name                                                       |
+| `config`              | HCL Block | Connector configuration                                              |
+| `config_sensitive`    | HCL Block | Sensitive connector configuration. Will be masked in output.         |
+
 ## Developing
 
 0. [Install go][install-go]

bin/test

@@ -4,6 +4,12 @@ set -ex
 
 go build
 mv terraform-provider-kafka-connect ~/.terraform.d/plugins/terraform-provider-kafka-connect
+docker-compose down
+docker-compose up -d
+sleep 60
 cd examples
+rm -rf .terraform terraform.tfstate*
 terraform init
 terraform plan
+terraform apply -auto-approve
+docker-compose down

connect/provider.go

@@ -17,6 +17,16 @@ func Provider() terraform.ResourceProvider {
 				Optional:    true,
 				DefaultFunc: schema.EnvDefaultFunc("KAFKA_CONNECT_URL", ""),
 			},
+			"basic_auth_username": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("KAFKA_CONNECT_BASIC_AUTH_USERNAME", ""),
+			},
+			"basic_auth_password": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("KAFKA_CONNECT_BASIC_AUTH_PASSWORD", ""),
+			},
 		},
 		ConfigureFunc: providerConfigure,
 		ResourcesMap: map[string]*schema.Resource{
@@ -31,6 +41,10 @@ func providerConfigure(d *schema.ResourceData) (interface{}, error) {
 	log.Printf("[INFO] Initializing KafkaConnect client")
 	addr := d.Get("url").(string)
 	c := kc.NewClient(addr)
-
+	user := d.Get("basic_auth_username").(string)
+	pass := d.Get("basic_auth_password").(string)
+	if user != "" && pass != "" {
+		c.SetBasicAuth(user, pass)
+	}
 	return c, nil
 }

connect/resource_kafka_connector.go

@@ -3,7 +3,6 @@ package connect
 import (
 	"fmt"
 	"log"
-	"time"
 
 	"github.com/hashicorp/terraform/helper/schema"
 	kc "github.com/ricardo-ch/go-kafka-connect/lib/connectors"
@@ -52,19 +51,11 @@ func setNameFromID(d *schema.ResourceData, meta interface{}) ([]*schema.Resource
 }
 
 func connectorCreate(d *schema.ResourceData, meta interface{}) error {
-	c := meta.(kc.Client)
+	c := meta.(kc.HighLevelClient)
 	name := nameFromRD(d)
 
 	config, sensitiveCache := configFromRD(d)
-	if !kc.TryUntil(
-		func() bool {
-			_, err := c.GetAll()
-			return err == nil
-		},
-		5*time.Minute,
-	) {
-		return fmt.Errorf("timed out trying to connect to kafka-connect server at %s", c.URL)
-	}
+
 	req := kc.CreateConnectorRequest{
 		ConnectorRequest: kc.ConnectorRequest{
 			Name: name,
@@ -87,7 +78,7 @@ func connectorCreate(d *schema.ResourceData, meta interface{}) error {
 }
 
 func connectorDelete(d *schema.ResourceData, meta interface{}) error {
-	c := meta.(kc.Client)
+	c := meta.(kc.HighLevelClient)
 
 	name := nameFromRD(d)
 	req := kc.ConnectorRequest{
@@ -105,7 +96,7 @@ func connectorDelete(d *schema.ResourceData, meta interface{}) error {
 }
 
 func connectorUpdate(d *schema.ResourceData, meta interface{}) error {
-	c := meta.(kc.Client)
+	c := meta.(kc.HighLevelClient)
 
 	name := nameFromRD(d)
 
@@ -135,7 +126,7 @@ func connectorUpdate(d *schema.ResourceData, meta interface{}) error {
 }
 
 func connectorRead(d *schema.ResourceData, meta interface{}) error {
-	c := meta.(kc.Client)
+	c := meta.(kc.HighLevelClient)
 
 	config, sensitiveCache := configFromRD(d)
 	name := d.Get("name").(string)
@@ -166,7 +157,7 @@ func connectorRead(d *schema.ResourceData, meta interface{}) error {
 // The first is intended to be passed to CreateConnectorRequest
 // The second is intended to preserve knowledge of which keys are sensitive information in the incoming
 // ConnectorResponse.Config
-func configFromRD(d *schema.ResourceData) (map[string]string, map[string]string) {
+func configFromRD(d *schema.ResourceData) (map[string]interface{}, map[string]interface{}) {
 	cfg := mapFromRD(d, "config")
 	scfg := mapFromRD(d, "config_sensitive")
 	config := combineMaps(cfg, scfg)
@@ -177,18 +168,13 @@ func nameFromRD(d *schema.ResourceData) string {
 	return d.Get("name").(string)
 }
 
-func mapFromRD(d *schema.ResourceData, key string) map[string]string {
-	mapToBe := d.Get(key).(map[string]interface{})
-	realMap := make(map[string]string)
-	for k, v := range mapToBe {
-		realMap[k] = v.(string)
-	}
-	return realMap
+func mapFromRD(d *schema.ResourceData, key string) map[string]interface{} {
+	return d.Get(key).(map[string]interface{})
 }
 
 // if there are duplicate keys this will always take the kv from second!!!
-func combineMaps(first map[string]string, second map[string]string) map[string]string {
-	union := make(map[string]string)
+func combineMaps(first map[string]interface{}, second map[string]interface{}) map[string]interface{} {
+	union := make(map[string]interface{})
 	for k, v := range first {
 		union[k] = v
 	}
@@ -198,7 +184,7 @@ func combineMaps(first map[string]string, second map[string]string) map[string]s
 	return union
 }
 
-func removeSecondKeysFromFirst(first map[string]string, second map[string]string) map[string]string {
+func removeSecondKeysFromFirst(first map[string]interface{}, second map[string]interface{}) map[string]interface{} {
 	for k := range second {
 		delete(first, k)
 	}

connect/resource_kafka_connector_test.go

@@ -49,7 +49,7 @@ func testResourceConnector_initialCheck(s *terraform.State) error {
 		return fmt.Errorf("id doesn't match name")
 	}
 
-	client := testProvider.Meta().(kc.Client)
+	client := testProvider.Meta().(kc.HighLevelClient)
 
 	c, err := client.GetConnector(kc.ConnectorRequest{Name: "sqlite-sink"})
 	if err != nil {
@@ -66,7 +66,7 @@ func testResourceConnector_initialCheck(s *terraform.State) error {
 }
 
 func testResourceConnector_updateCheck(s *terraform.State) error {
-	client := testProvider.Meta().(kc.Client)
+	client := testProvider.Meta().(kc.HighLevelClient)
 
 	c, err := client.GetConnector(kc.ConnectorRequest{Name: "sqlite-sink"})
 	if err != nil {

docker-compose.yaml

@@ -49,3 +49,14 @@ services:
       CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
       CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
       CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
+  nginx:
+    image: xscys/nginx-sidecar-basic-auth
+    ports:
+      - 8087:8087
+    depends_on:
+      - kafka-connect
+    environment:
+      FORWARD_HOST: kafka-connect
+      FORWARD_PORT: 8083
+      BASIC_AUTH_USERNAME: testuser
+      BASIC_AUTH_PASSWORD: testpassword

examples/main.tf

@@ -19,3 +19,29 @@ resource "kafka-connect_connector" "sqlite-sink" {
     "connection.password" = "this-should-never-appear-unmasked"
   }
 }
+
+provider "kafka-connect" {
+  alias = "with-basic-auth"
+  url = "http://localhost:8087"
+  basic_auth_username = "testuser"
+  basic_auth_password = "testpassword"
+}
+
+resource "kafka-connect_connector" "sqlite-sink-with-auth" {
+  provider = kafka-connect.with-basic-auth
+  name = "sqlite-sink-with-auth"
+
+  config = {
+    "name"            = "sqlite-sink-with-auth"
+    "connector.class" = "io.confluent.connect.jdbc.JdbcSinkConnector"
+    "tasks.max"       = 1
+    "topics"          = "orders"
+    "connection.url"  = "jdbc:sqlite:test.db"
+    "auto.create"     = "true"
+    "connection.user" = "admin"
+  }
+
+  config_sensitive = {
+    "connection.password" = "this-should-never-appear-unmasked"
+  }
+}

go.mod

@@ -4,7 +4,8 @@ go 1.12
 
 require (
 	github.com/hashicorp/terraform v0.12.1
-	github.com/ricardo-ch/go-kafka-connect v0.0.0-20180209135343-132b7b7ad380
+	github.com/ricardo-ch/go-kafka-connect v0.0.0-20200403115642-f7b66cb04ed7
+	gopkg.in/resty.v1 v1.12.0 // indirect
 )
 
 replace git.apache.org/thrift.git => github.com/apache/thrift v0.0.0-20180902110319-2566ecd5d999

go.sum

@@ -226,6 +226,8 @@ github.com/mitchellh/panicwrap v0.0.0-20190213213626-17011010aaa4/go.mod h1:YYMf
 github.com/mitchellh/prefixedio v0.0.0-20190213213902-5733675afd51/go.mod h1:kB1naBgV9ORnkiTVeyJOI1DavaJkG4oNIq0Af6ZVKUo=
 github.com/mitchellh/reflectwalk v1.0.0 h1:9D+8oIskB4VJBN5SFlmc27fSlIBZaov1Wpk/IfikLNY=
 github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
+github.com/mmajis/go-kafka-connect v0.0.0-20200328184024-6284b2164d53 h1:Yk+eTGgUgsl3mdjSeEEQ0R5pP/4ytJ7gK/1JY/J1BV8=
+github.com/mmajis/go-kafka-connect v0.0.0-20200328184024-6284b2164d53/go.mod h1:qmte3Nxjt2qkpVCRCzqKmD5HVqHolDYSiJ+I3dYf2wo=
 github.com/neelance/astrewrite v0.0.0-20160511093645-99348263ae86/go.mod h1:kHJEU3ofeGjhHklVoIGuVj85JJwZ6kWPaJwCIxgnFmo=
 github.com/neelance/sourcemap v0.0.0-20151028013722-8c68805598ab/go.mod h1:Qr6/a/Q4r9LP1IltGz7tA7iOK1WonHEYhu1HRBA7ZiM=
 github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d/go.mod h1:YUTz3bUH2ZwIWBy3CJBeOBEugqcmXREj14T+iG/4k4U=
@@ -252,6 +254,10 @@ github.com/ricardo-ch/go-kafka-connect v0.0.0-20180209135343-132b7b7ad380 h1:aNt
 github.com/ricardo-ch/go-kafka-connect v0.0.0-20180209135343-132b7b7ad380/go.mod h1:1wqdL63J2nVaqWpJuc1SH1nc+yuEYZrYI6JpxGhY3F4=
 github.com/ricardo-ch/go-kafka-connect v0.0.0-20190603085745-132b7b7ad380a6c860fecc0ca9b0a2af9bf45471 h1:72zdRhATkqj0AF2EgQpXK3rL77+hGy7rGNmzHNDXB7U=
 github.com/ricardo-ch/go-kafka-connect v0.0.0-20190603085745-132b7b7ad380a6c860fecc0ca9b0a2af9bf45471/go.mod h1:1wqdL63J2nVaqWpJuc1SH1nc+yuEYZrYI6JpxGhY3F4=
+github.com/ricardo-ch/go-kafka-connect v0.0.0-20190603085745-7ed69492c725 h1:GEYixMvGK8NsbWtTe+pMpYeqA7taxTWRTuurKmesv3M=
+github.com/ricardo-ch/go-kafka-connect v0.0.0-20190603085745-7ed69492c725/go.mod h1:1wqdL63J2nVaqWpJuc1SH1nc+yuEYZrYI6JpxGhY3F4=
+github.com/ricardo-ch/go-kafka-connect v0.0.0-20200403115642-f7b66cb04ed7 h1:0X6qZzZMIBLjw5GQKgBQiYJYozAcOYCTQqBv9tdwn78=
+github.com/ricardo-ch/go-kafka-connect v0.0.0-20200403115642-f7b66cb04ed7/go.mod h1:1wqdL63J2nVaqWpJuc1SH1nc+yuEYZrYI6JpxGhY3F4=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
@@ -337,6 +343,7 @@ golang.org/x/net v0.0.0-20181029044818-c44066c5c816/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20181106065722-10aee1819953/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181129055619-fae4c4e3ad76/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20181220203305-927f97764cc3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
@@ -404,6 +411,8 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/cheggaaa/pb.v1 v1.0.27/go.mod h1:V/YB90LKu/1FcN3WVnfiiE5oMCibMjukxqG/qStrOgw=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
+gopkg.in/resty.v1 v1.12.0 h1:CuXP0Pjfw9rOuY6EP+UvtNvt5DSqHpIxILZKT/quCZI=
+gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=