NOISSUE - Introduce NATS Jetstream as Default ES (#1907)

* Refactor message broker implementation This commit refactors the Nats message broker implementation to include pubsub options. These changes include: - Adding `Option` func that takes in the URL and prefix - Implement `WithStream` option which can create a different stream for nats stream - Implement `WithExchange` option which can create a different exchaange for rabbitmq channel - Implement `WithPrefix` option which allows to you change the publisher prefix These changes improve the organization and readability of the codebase. Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Use redis URL to configure username, password and db Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Make event store configurable on dev deployment Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Fix adds options to messaging `PubSub` interface Adding options to PubSub interface allows the use of messaging package to do es. The changes in this commit ensure that the code handles errors properly and provides more informative error messages when encountering unexpected types. Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Add NATS event publisher implementation This commit adds the implementation of the NATS event publisher. The NATS event publisher is responsible for publishing events to a NATS messaging system. It uses the `messaging` package to interact with the messaging system. The implementation includes the following features: - Publishing events to NATS using the `Publish` method. - Marshaling events to JSON before publishing. - Setting the message subject and headers based on the event. - Handling errors during publishing. This implementation is built with the `!rabbitmq` build tag, which means it will only be compiled if the `rabbitmq` build tag is not present. The NATS event publisher is part of the Mainflux events package and provides support for the Mainflux NATS events source service functionality. Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Add RabbitMQ event publisher implementation This commit adds the implementation of the RabbitMQ event publisher. The RabbitMQ event publisher is responsible for publishing events to a RabbitMQ messaging system. It uses the `messaging` package to interact with the messaging system. The implementation includes the following features: - Publishing events to RabbitMQ using the `Publish` method. - Marshaling events to JSON before publishing. - Setting the message subject and headers based on the event. - Handling errors during publishing. This implementation is built with the `rabbitmq` build tag, which means it will only be compiled if the `rabbitmq` build tag is present. The RabbitMQ event publisher is part of the Mainflux events package and provides support for the Mainflux RabbitMQ events source service functionality. Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Add configurable implementation for events store This commit adds a new file `brokers_*.go` which contains the implementation for the different event store. The file includes functions for creating a new publisher and subscriber using different es store. This commit also includes an `init` function that logs a message indicating that the binary was built using the respective package as the events store. The purpose of this commit is to add support for alternative events store. Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Fix build flags Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Refactor Makefile and Semaphore configuration The Makefile has been refactored to include the `MF_ES_STORE_TYPE` tag in the `go build` command. Additionally, the Semaphore configuration has been updated to include a new task for compiling with Redis as the broker type. This commit addresses the need to compile the codebase with Redis as the event store type and includes the necessary changes in the Makefile and Semaphore configuration. Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Reduced due to memory on testing Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Fix tests for es Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Fix grammar Co-authored-by: Sammy Kerata Oina <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Fix linting Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * feat(docker): update environment variables for message broker The commit updates the environment variable `MF_ES_STORE_TYPE` in the `docker/.env` file. The variable is changed from `${MF_MQTT_BROKER_TYPE}` to `${MF_MESSAGE_BROKER_TYPE}` to accurately reflect the type of message broker being used. This change ensures that the correct message broker is configured for the Event Store. Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * feat: Update docker environment variables - Removed the unused MF_ES_URL variable in the .env file - Updated the MF_ES_STORE_TYPE and MF_ES_STORE_URL variables in the .env file to match the MF_MESSAGE_BROKER_TYPE and MF_NATS_URL variables respectively Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> * Fix after rebase Signed-off-by: rodneyosodo <[email protected]> * Add godocs for option parameters for brokers Signed-off-by: rodneyosodo <[email protected]> * pass by value exchange and prefix names Signed-off-by: rodneyosodo <[email protected]> * Rename option functions Signed-off-by: rodneyosodo <[email protected]> * move variables to constants Signed-off-by: rodneyosodo <[email protected]> * fix: option example comment Signed-off-by: rodneyosodo <[email protected]> --------- Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: Rodney Osodo <[email protected]> Signed-off-by: rodneyosodo <[email protected]> Co-authored-by: Sammy Kerata Oina <[email protected]>
dborovcanin · Oct 24, 2023 · 43a263d · 43a263d
1 parent 059b954
commit 43a263d
Show file tree

Hide file tree

Showing 51 changed files with 1,915 additions and 181 deletions.
diff --git a/.semaphore/semaphore.yml b/.semaphore/semaphore.yml
@@ -19,13 +19,13 @@ blocks:
       jobs:
         - name: Setup Codecov
           commands:
-            - 'curl -Os https://uploader.codecov.io/latest/linux/codecov'
+            - "curl -Os https://uploader.codecov.io/latest/linux/codecov"
             - chmod +x codecov
             - ./codecov
             - cache store codecov ./codecov
         - name: Setup Golangci-lint
           commands:
-            - 'curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s v1.53.3'
+            - "curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s v1.53.3"
             - cache store linter ./bin/golangci-lint
       secrets:
         - name: codecov
@@ -50,10 +50,10 @@ blocks:
 
             - go install google.golang.org/protobuf/cmd/protoc-gen-go@$PROTOC_GEN_VERSION
             - go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@$PROTOC_GRPC_VERSION
-                
+
             - export PATH=$PATH:/usr/local/bin/protoc
             - export PATH=$PATH:$HOME/go/bin
-             
+
             - |
               echo "Setting up Mainflux..."
               for p in $(ls ./*.pb.go); do
@@ -70,7 +70,7 @@ blocks:
                   exit 1
                 fi
               done
-            - |                
+            - |
               for p in $(ls pkg/messaging/*.pb.go); do
                 if ! cmp -s $p $p.tmp; then
                   echo "Proto file and generated Go file $p are out of sync!"
@@ -89,7 +89,7 @@ blocks:
           commands:
             - cd users
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Users
           commands:
             - cd users
@@ -110,7 +110,7 @@ blocks:
           commands:
             - make docker_users
       secrets:
-        - name: docker_hub                   
+        - name: docker_hub
   - name: Lint and Test Things
     run:
       when: "change_in(['things', 'cmd/things', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -123,7 +123,7 @@ blocks:
           commands:
             - cd things
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Things
           commands:
             - cd things
@@ -144,7 +144,7 @@ blocks:
           commands:
             - make docker_things
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test CoAP
     run:
       when: "change_in(['coap', 'cmd/coap', 'things/api/grpc', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -157,7 +157,7 @@ blocks:
           commands:
             - cd coap
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test CoAP
           commands:
             - cd coap
@@ -178,7 +178,7 @@ blocks:
           commands:
             - make docker_coap
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test HTTP
     run:
       when: "change_in(['http', 'cmd/http', 'things/api/grpc', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -191,7 +191,7 @@ blocks:
           commands:
             - cd http
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test HTTP
           commands:
             - cd http
@@ -213,7 +213,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/http:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test MQTT
     run:
       when: "change_in(['mqtt', 'cmd/mqtt', 'things/api/grpc', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -226,7 +226,7 @@ blocks:
           commands:
             - cd mqtt
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test MQTT
           commands:
             - cd mqtt
@@ -248,7 +248,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/mqtt:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test WS
     run:
       when: "change_in(['ws', 'cmd/ws', 'things/api/grpc', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -261,7 +261,7 @@ blocks:
           commands:
             - cd ws
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test WS
           commands:
             - cd ws
@@ -283,7 +283,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/ws:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test Bootstrap
     run:
       when: "change_in(['bootstrap', 'cmd/bootstrap','things/policies/postgres/policies.go', 'things/policies/api/grpc/client.go'])"
@@ -296,7 +296,7 @@ blocks:
           commands:
             - cd bootstrap
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Bootstrap
           commands:
             - cd bootstrap
@@ -318,7 +318,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/bootstrap:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test Certs
     run:
       when: "change_in(['certs', 'cmd/certs', 'things/api/grpc', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -331,7 +331,7 @@ blocks:
           commands:
             - cd certs
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Certs
           commands:
             - cd certs
@@ -353,7 +353,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/certs:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test Provision
     dependencies:
       - Setup
@@ -366,7 +366,7 @@ blocks:
           commands:
             - cd provision
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Provision
           commands:
             - cd provision
@@ -388,7 +388,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/http:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test Twins
     run:
       when: "change_in(['twins', 'cmd/twins', 'things/api/grpc', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -401,7 +401,7 @@ blocks:
           commands:
             - cd twins
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Twins
           commands:
             - cd twins
@@ -423,7 +423,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/twins:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test Readers
     run:
       when: "change_in(['/readers', 'cmd/cassandra-reader', 'cmd/influxdb-reader', 'cmd/mongodb-reader', 'cmd/postgres-reader', 'cmd/timescale-reader', 'things/api/grpc', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -436,7 +436,7 @@ blocks:
           commands:
             - cd readers
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Twins
           commands:
             - cd readers
@@ -462,7 +462,7 @@ blocks:
             - docker push mainflux/postgres-reader:latest
             - docker push mainflux/timescale-reader:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test Consumers
     run:
       when: "change_in(['consumers', 'cmd/cassandra-writer', 'cmd/influxdb-writer', 'cmd/mongodb-writer', 'cmd/postgres-writer', 'cmd/timescale-writer', 'cmd/smpp-notifier', 'cmd/smtp-notifier', 'things/api/grpc', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -475,7 +475,7 @@ blocks:
           commands:
             - cd consumers
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Consumers
           commands:
             - cd consumers
@@ -502,7 +502,7 @@ blocks:
             - docker push mainflux/timescale-writer:latest
             - docker push mainflux/smtp-notifier:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test CLI
     run:
       when: "change_in(['cli', 'cmd/cli'])"
@@ -515,7 +515,7 @@ blocks:
           commands:
             - cd cli
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test CLI
           commands:
             - cd cli
@@ -537,7 +537,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/cli:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test LoRa
     run:
       when: "change_in(['lora', 'cmd/lora'])"
@@ -550,7 +550,7 @@ blocks:
           commands:
             - cd lora
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test LoRa
           commands:
             - cd lora
@@ -572,7 +572,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/lora:latest
       secrets:
-        - name: docker_hub             
+        - name: docker_hub
   - name: Lint and Test OPC-UA
     run:
       when: "change_in(['/opcua', 'cmd/opcua', 'things/api/grpc', 'auth/service.go', 'auth/api/grpc/client.go'])"
@@ -585,7 +585,7 @@ blocks:
           commands:
             - cd opcua
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test OPC-UA
           commands:
             - cd opcua
@@ -607,7 +607,7 @@ blocks:
             - echo $DOCKER_TOKEN | docker login --username "$DOCKER_USERNAME" --password-stdin
             - docker push mainflux/opcua:latest
       secrets:
-        - name: docker_hub              
+        - name: docker_hub
   - name: Lint and Test Internal
     run:
       when: "change_in('/internal')"
@@ -620,7 +620,7 @@ blocks:
           commands:
             - cd internal
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Internal
           commands:
             - cd internal
@@ -640,7 +640,7 @@ blocks:
           commands:
             - cd logger
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Logger
           commands:
             - cd logger
@@ -660,7 +660,7 @@ blocks:
           commands:
             - cd pkg
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test PKG
           commands:
             - cd pkg
@@ -680,7 +680,7 @@ blocks:
           commands:
             - cd tools
             - cache restore linter
-            - './bin/golangci-lint run'
+            - "./bin/golangci-lint run"
         - name: Test Tools
           commands:
             - cd tools
@@ -697,3 +697,11 @@ blocks:
           commands:
             - MF_MESSAGE_BROKER_TYPE=rabbitmq make mqtt
 
+  - name: Compile Check For Redis
+    dependencies:
+      - Setup
+    task:
+      jobs:
+        - name: Compile For Redis
+          commands:
+            - MF_ES_STORE_TYPE=redis make mqtt
diff --git a/Makefile b/Makefile
@@ -35,10 +35,15 @@ else
     MF_MQTT_BROKER_TYPE=nats
 endif
 
+ifneq ($(MF_ES_STORE_TYPE),)
+    MF_ES_STORE_TYPE := $(MF_ES_STORE_TYPE)
+else
+    MF_ES_STORE_TYPE=nats
+endif
 
 define compile_service
 	CGO_ENABLED=$(CGO_ENABLED) GOOS=$(GOOS) GOARCH=$(GOARCH) GOARM=$(GOARM) \
-	go build -mod=vendor -tags $(MF_MESSAGE_BROKER_TYPE) -ldflags "-s -w \
+	go build -mod=vendor -tags $(MF_MESSAGE_BROKER_TYPE) --tags $(MF_ES_STORE_TYPE) -ldflags "-s -w \
 	-X 'github.com/mainflux/mainflux.BuildTime=$(TIME)' \
 	-X 'github.com/mainflux/mainflux.Version=$(VERSION)' \
 	-X 'github.com/mainflux/mainflux.Commit=$(COMMIT)'" \
@@ -226,7 +231,15 @@ else
 endif
 
 run: check_certs change_config
+ifeq ($(MF_ES_STORE_TYPE), redis)
+	sed -i "s/MF_ES_STORE_TYPE=.*/MF_ES_STORE_TYPE=redis/" docker/.env
+	sed -i "s/MF_ES_STORE_URL=.*/MF_ES_STORE_URL=$$\{MF_REDIS_URL}/" docker/.env
+	docker-compose -f docker/docker-compose.yml --profile $(DOCKER_PROFILE) --profile redis -p $(DOCKER_PROJECT) $(DOCKER_COMPOSE_COMMAND) $(args)
+else
+	sed -i "s,MF_ES_STORE_TYPE=.*,MF_ES_STORE_TYPE=$$\{MF_MESSAGE_BROKER_TYPE}," docker/.env
+	sed -i "s,MF_ES_STORE_URL=.*,MF_ES_STORE_URL=$$\{MF_$(shell echo ${MF_MESSAGE_BROKER_TYPE} | tr 'a-z' 'A-Z')_URL\}," docker/.env
 	docker-compose -f docker/docker-compose.yml --profile $(DOCKER_PROFILE) -p $(DOCKER_PROJECT) $(DOCKER_COMPOSE_COMMAND) $(args)
+endif
 
 run_addons: check_certs
 	$(call change_config)