Merge pull request #6 from snapp-incubator/fix/fix-bugs

fix: fix problems related to config and stream creation

README.md

@@ -26,7 +26,7 @@ You can deploy and use NATS Blackbox Exporter using Docker images or by building
 ### 1. Using Docker Image 📫
 You can use pre-built Docker images from GitHub Container Registry (GHCR):
 ```bash
-docker run -d -p 8080:8080 --name nats-blackbox-exporter -v ./config.yaml:/app/config.yaml:ro ghcr.io/snapp-incubator/nats-blackbox-exporter:<release-tag>
+docker run -d -p 8080:8080 --name nats-blackbox-exporter -v ./setting/config.yaml:/app/setting/config.yaml:ro ghcr.io/snapp-incubator/nats-blackbox-exporter:<release-tag>
 ```
 and then pass environment variables as needed.
 
@@ -44,7 +44,7 @@ The exporter will generate Prometheus metrics on the port specified in the confi
 - **SuccessCounter:** A `prometheus.CounterVec` that counts successful publishes and consumes.
 
 ## 🎨 Configuration
-You can check the list of parameters with default values in the [config.example.yaml](./config.example.yaml) file. The NATS Blackbox Exporter can be configured in three ways:
+You can check the list of parameters with default values in the [config.example.yaml](./setting/config.example.yaml) file. The NATS Blackbox Exporter can be configured in three ways:
 
 1. **Environment Variables:**
    Set the necessary environment variables before running the exporter.

internal/config/config.go

@@ -42,7 +42,7 @@ func New() Config {
 	}
 
 	// load configuration from file
-	if err := k.Load(file.Provider("config.yaml"), yaml.Parser()); err != nil {
+	if err := k.Load(file.Provider("setting/config.yaml"), yaml.Parser()); err != nil {
 		log.Printf("error loading config.yaml")
 	}
 

internal/config/default.go

@@ -16,6 +16,7 @@ func Default() Config {
 			Level: "debug",
 		},
 		NATS: natsclient.Config{
+			NewStreamAllow: true,
 			Stream: natsclient.Stream{
 				Name:    "stream",
 				Subject: "test",

internal/natsclient/config.go

@@ -3,6 +3,7 @@ package natsclient
 import "time"
 
 type Config struct {
+	NewStreamAllow         bool          `json:"new_stream_allow" koanf:"new_stream_allow"`
 	Stream                 Stream        `json:"stream,omitempty"            koanf:"stream"`
 	URL                    string        `json:"url,omitempty"            koanf:"url"`
 	PublishInterval        time.Duration `json:"publish_interval" koanf:"publish_interval"`

internal/natsclient/jetstream.go

@@ -1,7 +1,6 @@
 package natsclient
 
 import (
-	"fmt"
 	"time"
 
 	"github.com/nats-io/nats.go"
@@ -66,14 +65,25 @@ func (j *Jetstream) createJetstreamContext() {
 }
 
 func (j *Jetstream) createStream() {
-	fmt.Println([]string{j.config.Stream.Subject})
-	fmt.Println(j.config.Stream.Name)
-	_, err := j.jetstream.AddStream(&nats.StreamConfig{
-		Name:     j.config.Stream.Name,
-		Subjects: []string{j.config.Stream.Subject},
-	})
-	if err != nil {
-		j.logger.Panic("could not add stream", zap.Error(err))
+	_, err := j.jetstream.StreamInfo(j.config.Stream.Name)
+	if err == nil {
+		_, err = j.jetstream.UpdateStream(&nats.StreamConfig{
+			Name:     j.config.Stream.Name,
+			Subjects: []string{j.config.Stream.Subject},
+		})
+		if err != nil {
+			j.logger.Panic("could not add subject to existing stream", zap.Error(err))
+		}
+	} else if err == nats.ErrStreamNotFound && j.config.NewStreamAllow {
+		_, err = j.jetstream.AddStream(&nats.StreamConfig{
+			Name:     j.config.Stream.Name,
+			Subjects: []string{j.config.Stream.Subject},
+		})
+		if err != nil {
+			j.logger.Panic("could not add stream", zap.Error(err))
+		}
+	} else {
+		j.logger.Panic("could not add subject", zap.Error(err))
 	}
 }
 

config.example.yaml → setting/config.example.yaml

@@ -2,9 +2,10 @@ logger:
   level: "debug"
 
 nats:
+  new_stream_allow: true
   stream:         
-    - name: "stream"
-      subject: "test"   # optional
+    name: "stream"
+    subject: "test"   # optional
   url: "localhost:4222"
   publish_interval: 2s
   request_timeout: 50ms