Support updating some consumer properties

Also support user, pass, token and nkey auth options.
Few bug fixes in CA file support and code style updates

Also stop using the old KV interface, moved to nats.go
version

Signed-off-by: R.I.Pienaar <[email protected]>

Author: ripienaar

README.md

@@ -93,9 +93,15 @@ provider "jetstream" {
 
 ### Argument Reference
 
- * `servers` - Comma separated list of servers to connect to
- * `credentials` - (optional) path to the credentials file to use
- * `credential_data` - (optional) the credentials as a string
+ * `servers` - The list of servers to connect to in a comma seperated list
+ * `credentials` - (optional) Fully Qualified Path to a file holding NATS credentials.
+ * `credential_data` - (optional) The NATS credentials as a string, intended to use with data providers.
+ * `user` - (optional) Connects using a username, when no password is set this is assumed to be a Token.
+ * `password` - (optional) Connects using a password
+ * `nkey` - (optional) Connects using an nkey stored in a file
+ * `tls.ca_file` - (optional) Fully Qualified Path to a file containing Root CA (PEM format). Use when the server has certs signed by an unknown authority.
+ * `tls.ca_file_data` - (optional) The Root CA PEM as a string, intended to use with data providers. Use when the server has certs signed by an unknown authority.
+
 
 ## jetstream_stream
 
@@ -198,6 +204,7 @@ resource "jetstream_consumer" "ORDERS_NEW" {
  * `stream_id` - The name of the Stream that this consumer consumes
  * `stream_sequence` - (optional) The Stream Sequence that will be the first message delivered by this Consumer
  * `ratelimit` - (optional) The rate limit for delivering messages to push consumers, expressed in bits per second
+ * `headers_only` - (optional) When true no message bodies will be delivered only headers
 
 ## jetstream_kv_bucket
 

docs/index.md

@@ -37,6 +37,9 @@ resource "jetstream_consumer" "ORDERS_NEW" {
  * `servers` - The list of servers to connect to in a comma seperated list
  * `credentials` - (optional) Fully Qualified Path to a file holding NATS credentials.
  * `credential_data` - (optional) The NATS credentials as a string, intended to use with data providers.
+ * `user` - (optional) Connects using a username, when no password is set this is assumed to be a Token.
+ * `password` - (optional) Connects using a password
+ * `nkey` - (optional) Connects using an nkey stored in a file
  * `tls.ca_file` - (optional) Fully Qualified Path to a file containing Root CA (PEM format). Use when the server has certs signed by an unknown authority.
  * `tls.ca_file_data` - (optional) The Root CA PEM as a string, intended to use with data providers. Use when the server has certs signed by an unknown authority.
 

docs/resources/jetstream_consumer.md

@@ -1,6 +1,6 @@
 # jetstream_stream Resource
 
-The `jetstream_consumer` Resource creates or deletes JetStream Consumers on any Terraform managed Stream. Does not support editing consumers in place.
+The `jetstream_consumer` Resource creates, updates or deletes JetStream Consumers on any Terraform managed Stream. 
 
 ## Example Usage
 
@@ -41,3 +41,4 @@ resource "jetstream_consumer" "ORDERS_NEW" {
  * `heartbeat` - (optional) Enable heartbeat messages for push consumers, duration specified in seconds
  * `flow_control` - (optional) Enable flow control for push consumers
  * `max_waiting` - (optional) The number of pulls that can be outstanding on a pull consumer, pulls received after this is reached are ignored
+ * `headers_only` - (optional) When true no message bodies will be delivered only headers

go.mod

@@ -5,10 +5,10 @@ go 1.16
 require (
 	github.com/google/go-cmp v0.5.6
 	github.com/hashicorp/terraform-plugin-sdk v1.17.2
-	github.com/nats-io/jsm.go v0.0.26
+	github.com/nats-io/jsm.go v0.0.27-0.20211124133852-b1f2cc1af686
 	github.com/nats-io/jwt v1.2.2
-	github.com/nats-io/nats-server/v2 v2.4.1-0.20210907200628-874c79fe411f
-	github.com/nats-io/nats.go v1.12.0
+	github.com/nats-io/nats-server/v2 v2.6.6-0.20211122213926-f094918f35b8
+	github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc
 	github.com/xeipuuv/gojsonschema v1.2.0
 	github.com/zclconf/go-cty v1.8.4 // indirect
 	golang.org/x/mod v0.5.1 // indirect

go.sum

@@ -138,7 +138,6 @@ github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QD
 github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
-github.com/golang/snappy v0.0.3 h1:fHPg5GQYlCeLIPB9BZqMVR5nR9A+IM5zcgeTdjMYmLA=
 github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
@@ -244,8 +243,9 @@ github.com/kevinburke/ssh_config v0.0.0-20201106050909-4977a11b4351/go.mod h1:CT
 github.com/keybase/go-crypto v0.0.0-20161004153544-93f5b35093ba/go.mod h1:ghbZscTyKdM07+Fw3KSi0hcJm+AlEUWj8QLlPtijN/M=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.11.2/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
-github.com/klauspost/compress v1.13.4 h1:0zhec2I8zGnjWcKyLl6i3gPqKANCCn5e9xmviEEeX6s=
 github.com/klauspost/compress v1.13.4/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
+github.com/klauspost/compress v1.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc=
+github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
@@ -288,16 +288,17 @@ github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh
 github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
 github.com/mitchellh/reflectwalk v1.0.1 h1:FVzMWA5RllMAKIdUSC8mdWo3XtwoecrH79BY70sEEpE=
 github.com/mitchellh/reflectwalk v1.0.1/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
-github.com/nats-io/jsm.go v0.0.26 h1:Ocj9wy/tfGJ+lOaxyYwLv2skG5CFSLuhRlg7DhP3WQI=
-github.com/nats-io/jsm.go v0.0.26/go.mod h1:jeU4Spx3HMszhjbvCyQNoWKdIJaj8ResDalLHFysbew=
+github.com/nats-io/jsm.go v0.0.27-0.20211124133852-b1f2cc1af686 h1:3qE1c3jOiGTmFLTQF8M0eBKmj1nq+PGyxvmHFWhwbpU=
+github.com/nats-io/jsm.go v0.0.27-0.20211124133852-b1f2cc1af686/go.mod h1:zuH+RjhsJ/o9Dd1ArZxcEpu48oB3pdsIVLbz3gqen/c=
 github.com/nats-io/jwt v1.2.2 h1:w3GMTO969dFg+UOKTmmyuu7IGdusK+7Ytlt//OYH/uU=
 github.com/nats-io/jwt v1.2.2/go.mod h1:/xX356yQA6LuXI9xWW7mZNpxgF2mBmGecH+Fj34sP5Q=
-github.com/nats-io/jwt/v2 v2.0.3 h1:i/O6cmIsjpcQyWDYNcq2JyZ3/VTF8SJ4JWluI5OhpvI=
-github.com/nats-io/jwt/v2 v2.0.3/go.mod h1:VRP+deawSXyhNjXmxPCHskrR6Mq50BqpEI5SEcNiGlY=
-github.com/nats-io/nats-server/v2 v2.4.1-0.20210907200628-874c79fe411f h1:r+bnrlIkFeYHwqxvFGwHT3ajx3Yji6frxH+X+ZgGxeA=
-github.com/nats-io/nats-server/v2 v2.4.1-0.20210907200628-874c79fe411f/go.mod h1:TUAhMFYh1VISyY/D4WKJUMuGHg8yHtoUTuxkbiej1lc=
-github.com/nats-io/nats.go v1.12.0 h1:n0oZzK2aIZDMKuEiMKJ9qkCUgVY5vTAAksSXtLlz5Xc=
-github.com/nats-io/nats.go v1.12.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/jwt/v2 v2.2.0 h1:Yg/4WFK6vsqMudRg91eBb7Dh6XeVcDMPHycDE8CfltE=
+github.com/nats-io/jwt/v2 v2.2.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
+github.com/nats-io/nats-server/v2 v2.6.6-0.20211122213926-f094918f35b8 h1:UKhEhYEiZhmFabtgUyjsVEQdlfrvVrJkavxQ+++aSk4=
+github.com/nats-io/nats-server/v2 v2.6.6-0.20211122213926-f094918f35b8/go.mod h1:n8O5NeknIIQgQld7//20NnQpCe1o5xIjVFxzh7IIZ6Y=
+github.com/nats-io/nats.go v1.13.1-0.20211018182449-f2416a8b1483/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc h1:SHr4MUUZJ/fAC0uSm2OzWOJYsHpapmR86mpw7q1qPXU=
+github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=

jetstream/provider.go

@@ -32,6 +32,23 @@ func Provider() terraform.ResourceProvider {
 				Optional:    true,
 				Description: "The contents of the NATS 2.0 Credentials file to use",
 			},
+			"user": {
+				Type:          schema.TypeString,
+				Optional:      true,
+				Description:   "Connect using an username, used as token when no password is given",
+				ConflictsWith: []string{"nkey", "credentials", "credential_data"},
+			},
+			"password": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				Description: "Connect using a password",
+			},
+			"nkey": {
+				Type:          schema.TypeString,
+				Optional:      true,
+				Description:   "Connect using a NKEY seed stored in a file",
+				ConflictsWith: []string{"user", "credentials", "credential_data"},
+			},
 			"tls": {
 				Type:     schema.TypeSet,
 				MaxItems: 1,

jetstream/resource_jetstream_consumer.go

@@ -18,6 +18,7 @@ func resourceConsumer() *schema.Resource {
 		Create: resourceConsumerCreate,
 		Read:   resourceConsumerRead,
 		Delete: resourceConsumerDelete,
+		Update: resourceConsumerUpdate,
 		Importer: &schema.ResourceImporter{
 			State: schema.ImportStatePassthrough,
 		},
@@ -34,7 +35,7 @@ func resourceConsumer() *schema.Resource {
 				Type:        schema.TypeString,
 				Description: "Contains additional information about this consumer",
 				Optional:    true,
-				ForceNew:    true,
+				ForceNew:    false,
 			},
 			"durable_name": {
 				Type:         schema.TypeString,
@@ -47,7 +48,7 @@ func resourceConsumer() *schema.Resource {
 				Type:        schema.TypeString,
 				Description: "The subject where a Push-based consumer will deliver messages",
 				Optional:    true,
-				ForceNew:    true,
+				ForceNew:    false,
 			},
 			"stream_sequence": {
 				Type:         schema.TypeInt,
@@ -111,14 +112,14 @@ func resourceConsumer() *schema.Resource {
 				Description: "Number of seconds to wait for acknowledgement",
 				Default:     30,
 				Optional:    true,
-				ForceNew:    true,
+				ForceNew:    false,
 			},
 			"max_delivery": {
 				Type:        schema.TypeInt,
 				Description: "Maximum deliveries to attempt for each message",
 				Default:     -1,
 				Optional:    true,
-				ForceNew:    true,
+				ForceNew:    false,
 			},
 			"filter_subject": {
 				Type:        schema.TypeString,
@@ -141,7 +142,7 @@ func resourceConsumer() *schema.Resource {
 				Optional:     true,
 				Default:      0,
 				ValidateFunc: validation.IntBetween(0, 100),
-				ForceNew:     true,
+				ForceNew:     false,
 			},
 			"ratelimit": {
 				Type:         schema.TypeInt,
@@ -157,7 +158,7 @@ func resourceConsumer() *schema.Resource {
 				Optional:     true,
 				Default:      20000,
 				ValidateFunc: validation.IntAtLeast(0),
-				ForceNew:     true,
+				ForceNew:     false,
 			},
 			"heartbeat": {
 				Type:        schema.TypeInt,
@@ -178,9 +179,16 @@ func resourceConsumer() *schema.Resource {
 				Description:  "The number of pulls that can be outstanding on a pull consumer, pulls received after this is reached are ignored",
 				Optional:     true,
 				Default:      512,
-				ForceNew:     true,
+				ForceNew:     false,
 				ValidateFunc: validation.IntAtLeast(0),
 			},
+			"headers_only": {
+				Type:        schema.TypeBool,
+				Description: "When true no message bodies will be delivered only headers",
+				Optional:    true,
+				Default:     false,
+				ForceNew:    false,
+			},
 		},
 	}
 }
@@ -198,6 +206,7 @@ func consumerConfigFromResourceData(d *schema.ResourceData) (cfg api.ConsumerCon
 		MaxAckPending:   d.Get("max_ack_pending").(int),
 		FlowControl:     d.Get("flow_control").(bool),
 		Heartbeat:       time.Duration(d.Get("heartbeat").(int)) * time.Second,
+		HeadersOnly:     d.Get("headers_only").(bool),
 	}
 
 	if description, ok := d.GetOk("description"); ok {
@@ -256,6 +265,69 @@ func consumerConfigFromResourceData(d *schema.ResourceData) (cfg api.ConsumerCon
 	return cfg, nil
 }
 
+func resourceConsumerUpdate(d *schema.ResourceData, m interface{}) error {
+	stream := d.Get("stream_id").(string)
+	if stream == "" {
+		return fmt.Errorf("cannot determine stream name for update")
+	}
+
+	durable := d.Get("durable_name").(string)
+	if durable == "" {
+		return fmt.Errorf("cannot determine durable name for update")
+	}
+
+	nc, mgr, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	if err != nil {
+		return err
+	}
+	defer nc.Close()
+
+	known, err := mgr.IsKnownConsumer(stream, durable)
+	if err != nil {
+		return err
+	}
+	if !known {
+		d.SetId("")
+		return nil
+	}
+
+	cons, err := mgr.LoadConsumer(stream, durable)
+	if err != nil {
+		return err
+	}
+
+	cfg, err := consumerConfigFromResourceData(d)
+	if err != nil {
+		return err
+	}
+
+	s := strings.TrimSuffix(cons.SampleFrequency(), "%")
+	freq, err := strconv.Atoi(s)
+	if err != nil {
+		return fmt.Errorf("failed to parse consumer sampling configuration: %v", err)
+	}
+
+	opts := []jsm.ConsumerOption{
+		jsm.ConsumerDescription(cfg.Description),
+		jsm.AckWait(cfg.AckWait),
+		jsm.MaxDeliveryAttempts(cfg.MaxDeliver),
+		jsm.SamplePercent(freq),
+		jsm.MaxAckPending(uint(cfg.MaxAckPending)),
+		jsm.MaxWaiting(uint(cfg.MaxWaiting)),
+	}
+
+	if cfg.HeadersOnly {
+		opts = append(opts, jsm.DeliverHeadersOnly())
+	}
+
+	err = cons.UpdateConfiguration(opts...)
+	if err != nil {
+		return err
+	}
+
+	return resourceConsumerRead(d, m)
+}
+
 func resourceConsumerCreate(d *schema.ResourceData, m interface{}) error {
 	cfg, err := consumerConfigFromResourceData(d)
 	if err != nil {
@@ -333,6 +405,7 @@ func resourceConsumerRead(d *schema.ResourceData, m interface{}) error {
 	d.Set("flow_control", cons.FlowControl())
 	d.Set("max_waiting", cons.MaxWaiting())
 	d.Set("delivery_group", cons.DeliverGroup())
+	d.Set("headers_only", cons.IsHeadersOnly())
 
 	switch cons.DeliverPolicy() {
 	case api.DeliverAll: