feat: update client lifecycle, recreate if it's have a stale connection

README.md

@@ -13,8 +13,8 @@ To perform local testing run the following commands:
 
 1. `docker compose up -d`
 2. `docker compose exec vault ash`
-3. `vault secrets enable -path=kafka vault-plugin-secrets-kafka`
-4. `vault write kafka/config  username=root password=rootpassword bootstrap_servers=kafka:29092`
+3. `vault secrets enable -path=kafka kafka`
+4. `vault write kafka/config username=root password=rootpassword bootstrap_servers=kafka:29092`
 5. `vault write -force kafka/role/svc`
 6. `vault read kafka/creds/svc` (should successfully create a user in the kafka
    cluster)
@@ -25,10 +25,10 @@ To perform local testing run the following commands:
 - [x] Write first working prototype
 - [x] Remove dependency on confluent-kafka go plugin (it's using C dependencies
       and because of this we can't perform statically linked builds)
-- [ ] Configure basic CI/CD flow
+- [x] Configure basic CI/CD flow
+- [x] Revisit kafka client lifecycle
+- [x] Improve credential revocation during service disruption events
 - [ ] Select SCRAM-SHA version on the plugin config level
 - [ ] Specify TLS certificates and CA on the plugin config level
-- [ ] Revisit kafka client lifecycle
-- [ ] Improve credential revocation during service disruption events
 - [ ] Revisit tests
 - [ ] Revisit acceptance tests

backend.go

@@ -82,13 +82,24 @@ func (b *kafkaBackend) invalidate(ctx context.Context, key string) {
 func (b *kafkaBackend) getClient(ctx context.Context, s logical.Storage) (*kafkaClient, error) {
 	b.lock.RLock()
 	unlockFunc := b.lock.RUnlock
+	clientAlive := true
 	defer func() { unlockFunc() }()
 
 	if b.client != nil {
-		return b.client, nil
+		_, _, err := b.client.DescribeCluster()
+		if err == nil {
+			return b.client, nil
+		}
+		clientAlive = false
 	}
 
 	b.lock.RUnlock()
+
+	if !clientAlive {
+		b.client.Close()
+		b.reset()
+	}
+
 	b.lock.Lock()
 	unlockFunc = b.lock.Unlock
 

client.go

@@ -70,9 +70,6 @@ func newClient(config *kafkaConfig) (*kafkaClient, error) {
 	brokers := strings.Split(config.BootstrapServers, ",")
 
 	kafkaConf := sarama.NewConfig()
-	kafkaConf.Producer.Retry.Max = 1
-	kafkaConf.Producer.RequiredAcks = sarama.WaitForAll
-	kafkaConf.Producer.Return.Successes = true
 	kafkaConf.Version = sarama.V3_5_1_0
 	kafkaConf.ClientID = "vault_sasl_scram_client"
 	kafkaConf.Metadata.Full = true

docker-compose.yaml

@@ -55,17 +55,20 @@ services:
     image: docker.io/golang:1.22.2-alpine
     entrypoint: /bin/ash
     environment:
-      GIT_TAG_NAME: snapshot
+      APP_VERSION: 0.0.1
+      APP_COMMIT: next
     command:
       - -c
       - |
-        # apk add git gcc libc-dev
         cd /app
-        rm -rf docker/vault/plugins/vault-plugin-secrets-kafka
-        CGO_ENABLED=0 go build -tags musl -ldflags "-s -w -X version.version=$${GIT_TAG_NAME}" -o docker/vault/plugins/vault-plugin-secrets-kafka cmd/vault-plugin-secrets-kafka/main.go
+        rm -rf /vault/plugins/kafka
+        mkdir -p /vault/plugins/
+        mkdir -p /vault/config/tls/
+        CGO_ENABLED=0 go build -tags musl -ldflags "-s -w -X version.version=$${APP_VERSION} -X version.commit=$${APP_COMMIT} -X version.date=$$(date -Iseconds)" -o /vault/plugins/kafka cmd/vault-plugin-secrets-kafka/main.go
     volumes:
-      - ./:/app
-      - cache:/go/pkg/mod
+      - ./:/app:ro
+      - vault-config:/vault:rw
+      - builder-cache:/go/pkg/mod
 
   kafka-provisioner:
     hostname: kafka-provisioner
@@ -149,9 +152,8 @@ services:
       VAULT_DEV_LISTEN_ADDRESS: 0.0.0.0:8200
       VAULT_DEV_ROOT_TOKEN_ID: roottoken
     volumes:
-      - vault-data:/var/lib/vault:rw
+      - vault-config:/vault:rw
       - kafka-secrets:/secrets:ro
-      - ./docker/vault:/vault:rw
     cap_add:
       - IPC_LOCK
     ports:
@@ -179,18 +181,11 @@ services:
       kafka:
         condition: service_healthy
 
-  # vault-provisioner:
-  #   hostname: vault-provisioner
-  #   image: docker.io/hashicorp/terraform:1.7.5
-  #   depends_on:
-  #     vault:
-  #       condition: service_healthy
-
 networks:
   private: {}
 
 volumes:
+  builder-cache: {}
   kafka-data: {}
   kafka-secrets: {}
-  vault-data: {}
-  cache: {}
+  vault-config: {}

kafka_credentials.go

@@ -53,7 +53,7 @@ func deleteCredential(ctx context.Context, c *kafkaClient, username string) erro
 		},
 	})
 	if err != nil {
-		return fmt.Errorf("error creating Kafka credentials: %w", err)
+		return fmt.Errorf("error revoking Kafka credentials: %w", err)
 	}
 
 	return nil
@@ -76,7 +76,7 @@ func (b *kafkaBackend) credentialRevoke(ctx context.Context, req *logical.Reques
 	}
 
 	if err := deleteCredential(ctx, client, username); err != nil {
-		return nil, fmt.Errorf("error revoking user credentials: %w", err)
+		return nil, err
 	}
 
 	return nil, nil

version/version.go

@@ -4,6 +4,7 @@ var (
 	version = "0.0.1"
 	commit  = "snapshot"
 	date    = "unknown"
+	builtBy = "local"
 )
 
 func Version() string {
@@ -18,6 +19,14 @@ func Date() string {
 	return date
 }
 
+func BuiltBy() string {
+	return builtBy
+}
+
 func Build() string {
 	return version + "+" + commit
 }
+
+func FullBuildInfo() string {
+	return version + "+" + commit + "+" + date + "+" + builtBy
+}