Merge pull request #128 from deinstapel/fix/consumer_account

Implement linked account userCredentials for consumer management

Author: wallyqs

controllers/jetstream/consumer.go

@@ -56,6 +56,7 @@ func (c *Controller) processConsumerObject(cns *apis.Consumer, jsmc jsmClient) (
 		remoteClientKey  string
 		remoteRootCA     string
 		accServers       []string
+		accUserCreds     string
 	)
 	if spec.Account != "" && c.opts.CRDConnect {
 		// Lookup the account using the REST client.
@@ -92,7 +93,28 @@ func (c *Controller) processConsumerObject(cns *apis.Consumer, jsmc jsmClient) (
 				}
 			}
 		}
-		// FIXME: Add support for UserCredentials for consumer.
+		// Lookup the UserCredentials.
+		if acc.Spec.Creds != nil {
+			secretName := acc.Spec.Creds.Secret.Name
+			secret, err := c.ki.Secrets(ns).Get(c.ctx, secretName, k8smeta.GetOptions{})
+			if err != nil {
+				return err
+			}
+
+			// Write the user credentials to the cache dir.
+			accDir := filepath.Join(c.cacheDir, ns, spec.Account)
+			if err := os.MkdirAll(accDir, 0755); err != nil {
+				return err
+			}
+			for k, v := range secret.Data {
+				if k == acc.Spec.Creds.File {
+					accUserCreds = filepath.Join(c.cacheDir, ns, spec.Account, k)
+					if err := os.WriteFile(filepath.Join(accDir, k), v, 0644); err != nil {
+						return err
+					}
+				}
+			}
+		}
 	}
 
 	defer func() {
@@ -134,7 +156,9 @@ func (c *Controller) processConsumerObject(cns *apis.Consumer, jsmc jsmClient) (
 			if remoteRootCA != "" {
 				opts = append(opts, nats.RootCAs(remoteRootCA))
 			}
-
+			if accUserCreds != "" {
+				opts = append(opts, nats.UserCredentials(accUserCreds))
+			}
 			if len(spec.TLS.RootCAs) > 0 {
 				opts = append(opts, nats.RootCAs(spec.TLS.RootCAs...))
 			}