Repository: absmach/magistrala (go)

Issue:
Bug: Docker Compose commands from README are incorrect
### What were you trying to achieve?

Run the project in Docker Compose environment

### What are the expected results?

Working and configured project

### What are the received results?

When running the command
```bash
docker-compose -f docker/docker-compose.yml up
```
An error was received: service nats is required by mqtt-adapter but is disabled. Can be enabled by profiles [nats_nats nats_rabbitmq]

### Steps To Reproduce

Run:
```bash
docker-compose -f docker/docker-compose.yml up
```

### In what environment did you encounter the issue?

Windows 11 Pro 22H2 22621.2792, go1.21.1 windows/amd64,

**Docker Compose** version v2.23.0-desktop.1

**docker version:**
**Client:**
 Cloud integration: v1.0.35+desktop.5
 Version:           24.0.6
 API version:       1.43
 Go version:        go1.20.7
 Git commit:        ed223bc
 Built:             Mon Sep  4 12:32:48 2023
 OS/Arch:           windows/amd64
 Context:           default

**Server:** Docker Desktop 4.25.2 (129061)
 Engine:
  Version:          24.0.6
  API version:      1.43 (minimum version 1.12)
  Go version:       go1.20.7
  Git commit:       1a79695
  Built:            Mon Sep  4 12:32:16 2023
  OS/Arch:          linux/amd64
  Experimental:     false
 containerd:
  Version:          1.6.22
  GitCommit:        8165feabfdfe38c65b599c4993d227328c231fca
 runc:
  Version:          1.1.8
  GitCommit:        v1.1.8-0-g82f18fe
 docker-init:
  Version:          0.19.0
  GitCommit:        de40ad0

### Additional information you deem important

Since `make run` is designed only for the linux/basch environment, there is no way to run it without adaptation. But judging by the README, everything should work.

After looking at the logic of the script in the `makefile`, I added to the command:

```bash
docker-compose -f docker/docker-compose.yml --profile nats_nats -p magistrala up
```
All services loaded and started.
Perhaps it's worth updating the README?

Pull request:
NOISSUE - Improve RE logging
<!-- Copyright (c) Abstract Machines
SPDX-License-Identifier: Apache-2.0 -->

<!--

Pull request title should be `MG-XXX - description` or `NOISSUE - description` where XXX is ID of the issue that this PR relate to.
Please review the [CONTRIBUTING.md](https://github.com/absmach/magistrala/blob/main/CONTRIBUTING.md) file for detailed contributing guidelines.

For Work In Progress Pull Requests, please use the Draft PR feature, see https://github.blog/2019-02-14-introducing-draft-pull-requests/ for further details.

For a timely review/response, please avoid force-pushing additional commits if your PR already received reviews or comments.

- Provide tests for your changes.
- Use descriptive commit messages.
- Comment your code where appropriate.
- Squash your commits
- Update any related documentation.
-->

# What type of PR is this?
This is an enhancement. It improves visibility within RE.
<!--This represents the type of PR you are submitting.

For example:
This is a bug fix because it fixes the following issue: #1234
This is a feature because it adds the following functionality: ...
This is a refactor because it changes the following functionality: ...
This is a documentation update because it updates the following documentation: ...
This is a dependency update because it updates the following dependencies: ...
This is an optimization because it improves the following functionality: ...
-->

## What does this do?
This pull request adds a logging layer to async RE rules execution.
<!--
Please provide a brief description of what this PR is intended to do.
Include List any changes that modify/break current functionality.
-->

## Which issue(s) does this PR fix/relate to?

<!--
For pull requests that relate or close an issue, please include them below.  We like to follow [Github's guidance on linking issues to pull requests](https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue).

For example having the text: "Resolves #1234" would connect the current pull request to issue 1234.  And when we merge the pull request, Github will automatically close the issue.
-->

There is no such issue.

## Have you included tests for your changes?

<!--If you have not included tests, please explain why.
For example:
Yes, I have included tests for my changes.
No, I have not included tests because I do not know how to.
-->
No.
## Did you document any new/modified feature?
N/A
<!--If you have not included documentation, please explain why.
For example:
Yes, I have updated the documentation for the new feature.
No, I have not updated the documentation because I do not know how to.
-->

### Notes
N/A
<!--Please provide any additional information you feel is important.-->


Patch:
diff --git a/cmd/re/main.go b/cmd/re/main.go
index b4fd8aeb1..e5e0c78db 100644
--- a/cmd/re/main.go
+++ b/cmd/re/main.go
@@ -190,7 +190,7 @@ func main() {
 	}
 	defer authnClient.Close()
 	logger.Info("AuthN  successfully connected to auth gRPC server " + authnClient.Secure())
-	errs := make(chan error, channBuffer)
+	runInfo := make(chan re.RunInfo, channBuffer)
 
 	domsGrpcCfg := grpcclient.Config{}
 	if err := env.ParseWithOptions(&domsGrpcCfg, env.Options{Prefix: envPrefixDomains}); err != nil {
@@ -233,7 +233,7 @@ func main() {
 	readersClient := grpcClient.NewReadersClient(client.Connection(), regrpcCfg.Timeout)
 	logger.Info("Readers gRPC client successfully connected to readers gRPC server " + client.Secure())
 
-	svc, err := newService(database, errs, msgSub, writersPub, alarmsPub, authz, ec, logger, readersClient)
+	svc, err := newService(database, runInfo, msgSub, writersPub, alarmsPub, authz, ec, logger, readersClient)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to create services: %s", err))
 		exitCode = 1
@@ -254,12 +254,8 @@ func main() {
 	}
 
 	go func() {
-		for err := range errs {
-			if err != nil {
-				logger.Warn("Error handling rule", slog.String("error", err.Error()))
-				continue
-			}
-			logger.Info("Handling rule completed successfully")
+		for info := range runInfo {
+			logger.LogAttrs(context.Background(), info.Level, info.Message, info.Details...)
 		}
 	}()
 
@@ -289,7 +285,7 @@ func main() {
 	}
 }
 
-func newService(db pgclient.Database, errs chan error, rePubSub messaging.PubSub, writersPub, alarmsPub messaging.Publisher, authz mgauthz.Authorization, ec email.Config, logger *slog.Logger, readersClient grpcReadersV1.ReadersServiceClient) (re.Service, error) {
+func newService(db pgclient.Database, runInfo chan re.RunInfo, rePubSub messaging.PubSub, writersPub, alarmsPub messaging.Publisher, authz mgauthz.Authorization, ec email.Config, logger *slog.Logger, readersClient grpcReadersV1.ReadersServiceClient) (re.Service, error) {
 	repo := repg.NewRepository(db)
 	idp := uuid.New()
 
@@ -298,7 +294,7 @@ func newService(db pgclient.Database, errs chan error, rePubSub messaging.PubSub
 		logger.Error(fmt.Sprintf("failed to configure e-mailing util: %s", err.Error()))
 	}
 
-	csvc := re.NewService(repo, errs, idp, rePubSub, writersPub, alarmsPub, re.NewTicker(time.Second*30), emailerClient, readersClient)
+	csvc := re.NewService(repo, runInfo, idp, rePubSub, writersPub, alarmsPub, re.NewTicker(time.Second*30), emailerClient, readersClient)
 	csvc, err = middleware.AuthorizationMiddleware(csvc, authz)
 	if err != nil {
 		return nil, err
diff --git a/re/handlers.go b/re/handlers.go
index f21d095b6..0d2a24109 100644
--- a/re/handlers.go
+++ b/re/handlers.go
@@ -5,6 +5,7 @@ package re
 
 import (
 	"context"
+	"log/slog"
 	"strconv"
 	"time"
 
@@ -37,14 +38,14 @@ func (re *re) Handle(msg *messaging.Message) error {
 
 	for _, r := range page.Rules {
 		go func(ctx context.Context) {
-			re.errors <- re.process(ctx, r, msg)
+			re.runInfo <- re.process(ctx, r, msg)
 		}(ctx)
 	}
 
 	return nil
 }
 
-func (re *re) process(ctx context.Context, r Rule, msg *messaging.Message) error {
+func (re *re) process(ctx context.Context, r Rule, msg *messaging.Message) RunInfo {
 	l := lua.NewState()
 	defer l.Close()
 	preload(l)
@@ -59,38 +60,37 @@ func (re *re) process(ctx context.Context, r Rule, msg *messaging.Message) error
 	l.SetGlobal("aes_encrypt", l.NewFunction(luaEncrypt))
 	l.SetGlobal("aes_decrypt", l.NewFunction(luaDecrypt))
 
+	details := []slog.Attr{
+		slog.String("domain_id", r.DomainID),
+		slog.String("rule_id", r.ID),
+		slog.String("rule_name", r.Name),
+		slog.Time("time", time.Now().UTC()),
+	}
 	if err := l.DoString(r.Logic.Value); err != nil {
-		return err
+		return RunInfo{Level: slog.LevelError, Message: "failed to run rule logic" + err.Error(), Details: details}
 	}
 	// Get the last result.
 	result := l.Get(-1)
 	if result == lua.LNil {
-		return nil
+		return RunInfo{Level: slog.LevelWarn, Message: "rule with nil script result", Details: details}
 	}
 	// Converting Lua is an expensive operation, so
 	// don't do it if there are no outputs.
 	if len(r.Logic.Outputs) == 0 {
-		return nil
+		return RunInfo{Level: slog.LevelWarn, Message: "rule with no output channels", Details: details}
 	}
 	var err error
 	res := convertLua(result)
 	for _, o := range r.Logic.Outputs {
 		// If value is false, don't run the follow-up.
 		if v, ok := res.(bool); ok && !v {
-			return nil
+			return RunInfo{Level: slog.LevelInfo, Message: err.Error(), Details: details}
 		}
 		if e := re.handleOutput(ctx, o, r, msg, res); e != nil {
 			err = errors.Wrap(e, err)
 		}
 	}
-	return err
-}
-
-func (re *re) processReportConfig(ctx context.Context, cfg ReportConfig) error {
-	if _, err := re.generateReport(ctx, cfg, EmailReport); err != nil {
-		return err
-	}
-	return nil
+	return RunInfo{Level: slog.LevelInfo, Message: "rule processed successfully", Details: details}
 }
 
 func (re *re) handleOutput(ctx context.Context, o ScriptOutput, r Rule, msg *messaging.Message, val interface{}) error {
@@ -125,14 +125,19 @@ func (re *re) StartScheduler(ctx context.Context) error {
 
 			page, err := re.repo.ListRules(ctx, pm)
 			if err != nil {
-				re.errors <- err
+				re.runInfo <- RunInfo{
+					Level:   slog.LevelError,
+					Message: "failed to list rules" + err.Error(),
+					Details: []slog.Attr{slog.Time("due", due)},
+				}
+
 				continue
 			}
 
 			for _, r := range page.Rules {
 				go func(rule Rule) {
 					if _, err := re.repo.UpdateRuleDue(ctx, rule.ID, rule.Schedule.NextDue()); err != nil {
-						re.errors <- err
+						re.runInfo <- RunInfo{Level: slog.LevelError, Message: "falied to update rule due" + err.Error(), Details: []slog.Attr{slog.Time("time", time.Now().UTC())}}
 						return
 					}
 
@@ -142,23 +147,42 @@ func (re *re) StartScheduler(ctx context.Context) error {
 						Protocol: protocol,
 						Created:  due.Unix(),
 					}
-					re.errors <- re.process(ctx, rule, msg)
+					re.runInfo <- re.process(ctx, rule, msg)
 				}(r)
 			}
+			// Reset due, it will reset the page meta as well.
+			due = time.Now().UTC()
 
 			reportConfigs, err := re.repo.ListReportsConfig(ctx, pm)
 			if err != nil {
-				re.errors <- err
+				re.runInfo <- RunInfo{
+					Level:   slog.LevelError,
+					Message: "fiald to list reports " + err.Error(),
+					Details: []slog.Attr{slog.Time("due", due)},
+				}
 				continue
 			}
 
 			for _, c := range reportConfigs.ReportConfigs {
 				go func(cfg ReportConfig) {
 					if _, err := re.repo.UpdateReportDue(ctx, cfg.ID, cfg.Schedule.NextDue()); err != nil {
-						re.errors <- err
+						re.runInfo <- RunInfo{Level: slog.LevelError, Message: "falied to update report due" + err.Error(), Details: []slog.Attr{slog.Time("time", time.Now().UTC())}}
 						return
 					}
-					re.errors <- re.processReportConfig(ctx, cfg)
+					_, err := re.generateReport(ctx, cfg, EmailReport)
+					info := RunInfo{
+						Details: []slog.Attr{
+							slog.String("domain_id", cfg.DomainID),
+							slog.String("report_id", cfg.ID),
+							slog.String("report_name", cfg.Name),
+							slog.Time("time", time.Now().UTC()),
+						},
+					}
+					if err != nil {
+						info.Level = slog.LevelError
+						info.Message = "failed to generate report" + err.Error()
+					}
+					re.runInfo <- info
 				}(c)
 			}
 		}
diff --git a/re/rule.go b/re/rule.go
index b23c0a3f0..19d4e25ae 100644
--- a/re/rule.go
+++ b/re/rule.go
@@ -5,6 +5,7 @@ package re
 
 import (
 	"encoding/json"
+	"log/slog"
 	"strings"
 	"time"
 
@@ -88,10 +89,16 @@ type Rule struct {
 	Logic         Script    `json:"logic"`
 	OutputChannel string    `json:"output_channel,omitempty"`
 	OutputTopic   string    `json:"output_topic,omitempty"`
-	Schedule      Schedule  `json:"schedule,omitempty"`
+	Schedule      Schedule  `json:"schedule"`
 	Status        Status    `json:"status"`
-	CreatedAt     time.Time `json:"created_at,omitempty"`
-	CreatedBy     string    `json:"created_by,omitempty"`
-	UpdatedAt     time.Time `json:"updated_at,omitempty"`
-	UpdatedBy     string    `json:"updated_by,omitempty"`
+	CreatedAt     time.Time `json:"created_at"`
+	CreatedBy     string    `json:"created_by"`
+	UpdatedAt     time.Time `json:"updated_at"`
+	UpdatedBy     string    `json:"updated_by"`
+}
+
+type RunInfo struct {
+	Level   slog.Level
+	Details []slog.Attr
+	Message string
 }
diff --git a/re/service.go b/re/service.go
index d60207826..b602efbc6 100644
--- a/re/service.go
+++ b/re/service.go
@@ -91,7 +91,7 @@ type Service interface {
 
 type re struct {
 	repo       Repository
-	errors     chan error
+	runInfo    chan RunInfo
 	idp        supermq.IDProvider
 	rePubSub   messaging.PubSub
 	writersPub messaging.Publisher
@@ -101,11 +101,11 @@ type re struct {
 	readers    grpcReadersV1.ReadersServiceClient
 }
 
-func NewService(repo Repository, errors chan (error), idp supermq.IDProvider, rePubSub messaging.PubSub, writersPub, alarmsPub messaging.Publisher, tck Ticker, emailer Emailer, readers grpcReadersV1.ReadersServiceClient) Service {
+func NewService(repo Repository, runInfo chan RunInfo, idp supermq.IDProvider, rePubSub messaging.PubSub, writersPub, alarmsPub messaging.Publisher, tck Ticker, emailer Emailer, readers grpcReadersV1.ReadersServiceClient) Service {
 	return &re{
 		repo:       repo,
 		idp:        idp,
-		errors:     errors,
+		runInfo:    runInfo,
 		rePubSub:   rePubSub,
 		writersPub: writersPub,
 		alarmsPub:  alarmsPub,


Test patch:
diff --git a/re/service_test.go b/re/service_test.go
index 997c48b23..7352c630b 100644
--- a/re/service_test.go
+++ b/re/service_test.go
@@ -51,18 +51,18 @@ var (
 	}
 )
 
-func newService(t *testing.T, errs chan error) (re.Service, *mocks.Repository, *pubsubmocks.PubSub, *mocks.Ticker) {
+func newService(t *testing.T, runInfo chan re.RunInfo) (re.Service, *mocks.Repository, *pubsubmocks.PubSub, *mocks.Ticker) {
 	repo := new(mocks.Repository)
 	mockTicker := new(mocks.Ticker)
 	idProvider := uuid.NewMock()
 	pubsub := pubsubmocks.NewPubSub(t)
 	readersSvc := new(readmocks.ReadersServiceClient)
 	e := new(mocks.Emailer)
-	return re.NewService(repo, errs, idProvider, pubsub, pubsub, pubsub, mockTicker, e, readersSvc), repo, pubsub, mockTicker
+	return re.NewService(repo, runInfo, idProvider, pubsub, pubsub, pubsub, mockTicker, e, readersSvc), repo, pubsub, mockTicker
 }
 
 func TestAddRule(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 	ruleName := namegen.Generate()
 	now := time.Now().Add(time.Hour)
 	cases := []struct {
@@ -137,7 +137,7 @@ func TestAddRule(t *testing.T) {
 }
 
 func TestViewRule(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	now := time.Now().Add(time.Hour)
 	cases := []struct {
@@ -195,7 +195,7 @@ func TestViewRule(t *testing.T) {
 }
 
 func TestUpdateRule(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	newName := namegen.Generate()
 	now := time.Now().Add(time.Hour)
@@ -280,7 +280,7 @@ func TestUpdateRule(t *testing.T) {
 }
 
 func TestListRules(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 	numRules := 50
 	now := time.Now().Add(time.Hour)
 	var rules []re.Rule
@@ -385,7 +385,7 @@ func TestListRules(t *testing.T) {
 }
 
 func TestRemoveRule(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	cases := []struct {
 		desc    string
@@ -425,7 +425,7 @@ func TestRemoveRule(t *testing.T) {
 }
 
 func TestEnableRule(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	now := time.Now()
 
@@ -484,7 +484,7 @@ func TestEnableRule(t *testing.T) {
 }
 
 func TestDisableRule(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	now := time.Now()
 
@@ -543,7 +543,7 @@ func TestDisableRule(t *testing.T) {
 }
 
 func TestHandle(t *testing.T) {
-	svc, repo, pubmocks, _ := newService(t, make(chan error))
+	svc, repo, pubmocks, _ := newService(t, make(chan re.RunInfo))
 	now := time.Now()
 	empty := ""
 
@@ -627,7 +627,7 @@ func TestHandle(t *testing.T) {
 }
 
 func TestAddReportConfig(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	cases := []struct {
 		desc    string
@@ -679,7 +679,7 @@ func TestAddReportConfig(t *testing.T) {
 }
 
 func TestViewReportConfig(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	cases := []struct {
 		desc    string
@@ -724,7 +724,7 @@ func TestViewReportConfig(t *testing.T) {
 }
 
 func TestUpdateReportConfig(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	newName := namegen.Generate()
 	now := time.Now().Add(time.Hour)
@@ -787,7 +787,7 @@ func TestUpdateReportConfig(t *testing.T) {
 }
 
 func TestListReportsConfig(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 	numConfigs := 50
 	now := time.Now().Add(time.Hour)
 	var configs []re.ReportConfig
@@ -893,7 +893,7 @@ func TestListReportsConfig(t *testing.T) {
 }
 
 func TestRemoveReportConfig(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	cases := []struct {
 		desc    string
@@ -933,7 +933,7 @@ func TestRemoveReportConfig(t *testing.T) {
 }
 
 func TestEnableReportConfig(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	cases := []struct {
 		desc    string
@@ -981,7 +981,7 @@ func TestEnableReportConfig(t *testing.T) {
 }
 
 func TestDisableReportConfig(t *testing.T) {
-	svc, repo, _, _ := newService(t, make(chan error))
+	svc, repo, _, _ := newService(t, make(chan re.RunInfo))
 
 	cases := []struct {
 		desc    string
@@ -1038,8 +1038,8 @@ func TestDisableReportConfig(t *testing.T) {
 
 func TestStartScheduler(t *testing.T) {
 	now := time.Now().Truncate(time.Minute)
-	errs := make(chan error)
-	svc, repo, _, ticker := newService(t, errs)
+	ri := make(chan re.RunInfo)
+	svc, repo, _, ticker := newService(t, ri)
 
 	ctxCases := []struct {
 		desc     string