feat: creates consumers if not present in NATS

The jetstream-controller its an operate that should ensure that a system
is in a given state. During our tests we found out that if a consumers
or stream are deleted via code or through `nats-box` they are not
created again by the controller.

From my understanding this is happening because we are checking if the
observed generation number and generation number match in the CRD, which
on this case (a manually delete consumer) will not change and thus not
trigger a recreation.

The following PR aims to resources (consumers and streams) being created
in NATS in case they are not found by the client but there is a CRD for it.

cmd/nats-boot-config/main.go

@@ -57,10 +57,14 @@ func main() {
 	if os.Getenv("DEBUG") == "true" {
 		log.SetLevel(log.DebugLevel)
 	}
-	formatter := &log.TextFormatter{
-		FullTimestamp: true,
+
+	if os.Getenv("LOGFORMAT") == "json" {
+		log.SetFormatter(&log.JSONFormatter{})
+	} else {
+		log.SetFormatter(&log.TextFormatter{
+			FullTimestamp: true,
+		})
 	}
-	log.SetFormatter(formatter)
 
 	controller := bootconfig.NewController(opts)
 	log.Infof("Starting NATS Server boot config v%s (%s, %s)", Version, GitInfo, BuildTime)

controllers/jetstream/consumer.go

@@ -172,7 +172,7 @@ func (c *Controller) processConsumerObject(cns *apis.Consumer, jsmc jsmClient) (
 		return err
 	}
 	updateOK := (consumerOK && !deleteOK && newGeneration)
-	createOK := (!consumerOK && !deleteOK && newGeneration)
+	createOK := (!consumerOK && !deleteOK)
 
 	switch {
 	case createOK:

controllers/jetstream/stream.go

@@ -184,7 +184,7 @@ func (c *Controller) processStreamObject(str *apis.Stream, jsmc jsmClient) (err
 		return err
 	}
 	updateOK := (strOK && !deleteOK && newGeneration)
-	createOK := (!strOK && !deleteOK && newGeneration)
+	createOK := (!strOK && !deleteOK)
 
 	switch {
 	case createOK: