Merge pull request #146 from nats-io/add-metadata

Add metadata support
nats-io · Sep 22, 2023 · 129291d · 129291d
2 parents 0e208aa + f4c520b
commit 129291d
Show file tree

Hide file tree

Showing 8 changed files with 102 additions and 34 deletions.
diff --git a/controllers/jetstream/consumer.go b/controllers/jetstream/consumer.go
@@ -433,6 +433,10 @@ func consumerSpecToOpts(spec apis.ConsumerSpec) ([]jsm.ConsumerOption, error) {
 		opts = append(opts, jsm.MaxDeliveryAttempts(spec.MaxDeliver))
 	}
 
+	if spec.Metadata != nil {
+		opts = append(opts, jsm.ConsumerMetadata(spec.Metadata))
+	}
+
 	return opts, nil
 }
 

diff --git a/controllers/jetstream/stream.go b/controllers/jetstream/stream.go
@@ -434,6 +434,10 @@ func createStream(ctx context.Context, c jsmClient, spec apis.StreamSpec) (err e
 		opts = append(opts, jsm.FirstSequence(uint64(spec.FirstSequence)))
 	}
 
+	if spec.Metadata != nil {
+		opts = append(opts, jsm.StreamMetadata(spec.Metadata))
+	}
+
 	_, err = c.NewStream(ctx, spec.Name, opts)
 	return err
 }
@@ -519,6 +523,10 @@ func updateStream(ctx context.Context, c jsmClient, spec apis.StreamSpec) (err e
 		config.Sources[i] = jss
 	}
 
+	if spec.Metadata != nil {
+		config.Metadata = spec.Metadata
+	}
+
 	switch spec.Compression {
 	case "s2":
 		config.Compression = api.S2Compression

diff --git a/deploy/crds.yml b/deploy/crds.yml
@@ -172,6 +172,11 @@ spec:
                           dest:
                             description: Destination subject.
                             type: string
+              metadata:
+                description: Additional Stream metadata.
+                type: object
+                additionalProperties:
+                  type: string
               servers:
                 description: A list of servers for creating stream
                 type: array
@@ -595,6 +600,11 @@ spec:
                 description: Force the consumer state to be kept in memory rather than inherit the setting from the stream.
                 type: boolean
                 default: false
+              metadata:
+                description: Additional Consumer metadata.
+                type: object
+                additionalProperties:
+                  type: string
               tls:
                 description: A client's TLS certs and keys.
                 type: object

diff --git a/pkg/jetstream/apis/jetstream/v1beta2/consumertypes.go b/pkg/jetstream/apis/jetstream/v1beta2/consumertypes.go
@@ -22,40 +22,41 @@ func (c *Consumer) GetSpec() interface{} {
 
 // ConsumerSpec is the spec for a Consumer resource
 type ConsumerSpec struct {
-	AckPolicy          string   `json:"ackPolicy"`
-	AckWait            string   `json:"ackWait"`
-	BackOff            []string `json:"backoff"`
-	Creds              string   `json:"creds"`
-	DeliverGroup       string   `json:"deliverGroup"`
-	DeliverPolicy      string   `json:"deliverPolicy"`
-	DeliverSubject     string   `json:"deliverSubject"`
-	Description        string   `json:"description"`
-	PreventDelete      bool     `json:"preventDelete"`
-	PreventUpdate      bool     `json:"preventUpdate"`
-	DurableName        string   `json:"durableName"`
-	FilterSubject      string   `json:"filterSubject"`
-	FilterSubjects     []string `json:"filterSubjects"`
-	FlowControl        bool     `json:"flowControl"`
-	HeadersOnly        bool     `json:"headersOnly"`
-	HeartbeatInterval  string   `json:"heartbeatInterval"`
-	MaxAckPending      int      `json:"maxAckPending"`
-	MaxDeliver         int      `json:"maxDeliver"`
-	MaxRequestBatch    int      `json:"maxRequestBatch"`
-	MaxRequestExpires  string   `json:"maxRequestExpires"`
-	MaxRequestMaxBytes int      `json:"maxRequestMaxBytes"`
-	MaxWaiting         int      `json:"maxWaiting"`
-	MemStorage         bool     `json:"memStorage"`
-	Nkey               string   `json:"nkey"`
-	OptStartSeq        int      `json:"optStartSeq"`
-	OptStartTime       string   `json:"optStartTime"`
-	RateLimitBps       int      `json:"rateLimitBps"`
-	ReplayPolicy       string   `json:"replayPolicy"`
-	Replicas           int      `json:"replicas"`
-	SampleFreq         string   `json:"sampleFreq"`
-	Servers            []string `json:"servers"`
-	StreamName         string   `json:"streamName"`
-	TLS                TLS      `json:"tls"`
-	Account            string   `json:"account"`
+	AckPolicy          string            `json:"ackPolicy"`
+	AckWait            string            `json:"ackWait"`
+	BackOff            []string          `json:"backoff"`
+	Creds              string            `json:"creds"`
+	DeliverGroup       string            `json:"deliverGroup"`
+	DeliverPolicy      string            `json:"deliverPolicy"`
+	DeliverSubject     string            `json:"deliverSubject"`
+	Description        string            `json:"description"`
+	PreventDelete      bool              `json:"preventDelete"`
+	PreventUpdate      bool              `json:"preventUpdate"`
+	DurableName        string            `json:"durableName"`
+	FilterSubject      string            `json:"filterSubject"`
+	FilterSubjects     []string          `json:"filterSubjects"`
+	FlowControl        bool              `json:"flowControl"`
+	HeadersOnly        bool              `json:"headersOnly"`
+	HeartbeatInterval  string            `json:"heartbeatInterval"`
+	MaxAckPending      int               `json:"maxAckPending"`
+	MaxDeliver         int               `json:"maxDeliver"`
+	MaxRequestBatch    int               `json:"maxRequestBatch"`
+	MaxRequestExpires  string            `json:"maxRequestExpires"`
+	MaxRequestMaxBytes int               `json:"maxRequestMaxBytes"`
+	MaxWaiting         int               `json:"maxWaiting"`
+	MemStorage         bool              `json:"memStorage"`
+	Nkey               string            `json:"nkey"`
+	OptStartSeq        int               `json:"optStartSeq"`
+	OptStartTime       string            `json:"optStartTime"`
+	RateLimitBps       int               `json:"rateLimitBps"`
+	ReplayPolicy       string            `json:"replayPolicy"`
+	Replicas           int               `json:"replicas"`
+	SampleFreq         string            `json:"sampleFreq"`
+	Servers            []string          `json:"servers"`
+	StreamName         string            `json:"streamName"`
+	TLS                TLS               `json:"tls"`
+	Account            string            `json:"account"`
+	Metadata           map[string]string `json:"metadata"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

diff --git a/pkg/jetstream/apis/jetstream/v1beta2/streamtypes.go b/pkg/jetstream/apis/jetstream/v1beta2/streamtypes.go
@@ -49,6 +49,7 @@ type StreamSpec struct {
 	SubjectTransform  *SubjectTransform `json:"subjectTransform"`
 	FirstSequence     uint64            `json:"firstSequence"`
 	Compression       string            `json:"compression"`
+	Metadata          map[string]string `json:"metadata"`
 	Retention         string            `json:"retention"`
 	Servers           []string          `json:"servers"`
 	Sources           []*StreamSource   `json:"sources"`

diff --git a/pkg/jetstream/apis/jetstream/v1beta2/zz_generated.deepcopy.go b/pkg/jetstream/apis/jetstream/v1beta2/zz_generated.deepcopy.go
diff --git a/pkg/jetstream/generated/applyconfiguration/jetstream/v1beta2/consumerspec.go b/pkg/jetstream/generated/applyconfiguration/jetstream/v1beta2/consumerspec.go
diff --git a/pkg/jetstream/generated/applyconfiguration/jetstream/v1beta2/streamspec.go b/pkg/jetstream/generated/applyconfiguration/jetstream/v1beta2/streamspec.go