Merge pull request #38 from nats-io/consumer_edit

Support updating some consumer properties

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: