feat: Prevent deletes in consumers and streams

Adding a `preventDelete` property for streams and consumers CRDS which
when set to false nack should be able to delete it, otherwise it should
just ignore that stream or consumer.

The use case for this is in a multi-cloud context when we have deploy
deployed in different clouds it becomes troublesome. For example, nack
on GKE might check that a stream CRD is missing and proceeds to delete
it, however that CRD exists in AWS and AKS and it should not be deleted
because its going to impact other applications.

controllers/jetstream/consumer.go

@@ -387,6 +387,11 @@ func deleteConsumer(ctx context.Context, c jsmClient, spec apis.ConsumerSpec) (e
 		}
 	}()
 
+	if !spec.PreventDelete {
+		fmt.Printf("Consumer %q is configured to preventDelete on stream %q:\n", stream, consumer)
+		return nil
+	}
+
 	var apierr jsmapi.ApiError
 	cn, err := c.LoadConsumer(ctx, stream, consumer)
 	if errors.As(err, &apierr) && apierr.NotFoundError() {

controllers/jetstream/stream.go

@@ -397,6 +397,11 @@ func deleteStream(ctx context.Context, c jsmClient, spec apis.StreamSpec) (err e
 		}
 	}()
 
+	if !spec.PreventDelete {
+		fmt.Printf("Stream %q is configured to preventDelete:\n", name)
+		return nil
+	}
+
 	var apierr jsmapi.ApiError
 	str, err := c.LoadStream(ctx, name)
 	if errors.As(err, &apierr) && apierr.NotFoundError() {

deploy/crds.yml

@@ -189,6 +189,10 @@ spec:
                   source:
                     type: string
                     description: Messages will be published from that subject to the destination subject.
+              preventDelete:
+                description: Indicates if a Stream can be deleted or not
+                type: boolean
+                default: false
           status:
             type: object
             properties:
@@ -551,6 +555,10 @@ spec:
                 description: Name of the account to which the Consumer belongs.
                 type: string
                 pattern: '^[^.*>]*$'
+              preventDelete:
+                description: Indicates if a Consumer can be deleted or not
+                type: boolean
+                default: false
           status:
             type: object
             properties:

pkg/jetstream/apis/jetstream/v1beta2/consumertypes.go

@@ -30,6 +30,7 @@ type ConsumerSpec struct {
 	DeliverPolicy      string   `json:"deliverPolicy"`
 	DeliverSubject     string   `json:"deliverSubject"`
 	Description        string   `json:"description"`
+	PreventDelete      bool     `json:"preventDelete"`
 	DurableName        string   `json:"durableName"`
 	FilterSubject      string   `json:"filterSubject"`
 	FlowControl        bool     `json:"flowControl"`

pkg/jetstream/apis/jetstream/v1beta2/streamtypes.go

@@ -25,6 +25,7 @@ type StreamSpec struct {
 	Account           string                `json:"account"`
 	Creds             string                `json:"creds"`
 	Description       string                `json:"description"`
+	PreventDelete     bool                  `json:"preventDelete"`
 	Discard           string                `json:"discard"`
 	DuplicateWindow   string                `json:"duplicateWindow"`
 	MaxAge            string                `json:"maxAge"`