MF-780 - Use Normalizer as a lib (#915)

* Use Normalizer as a lib

To normalize messages on the consumer side, Normalizer is moved
to the internal pkgs. Writers being message consumers are modified to
do message normalization instead of subscribing to normalized messages
subject.

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Fix logging middleware for readers and writers

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Remove normalizer interface

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Use Normalizer in writers

As we agreed on #919, we'll use normalizer as an interface and provide
the default SenML implementation. Because of that, Normalizer is removed
from `internal` and we'll use the project structure proposed in the
aforementioned issue.

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Fix tests

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Remove unused batch settings from influxDB reader

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Update docs

Move Normalizer service to `addons`.

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Rename channels input topic

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Update Noramlizer docs

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Remove commented code

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Update readers logging

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Update addons docker-compose files

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Update topcis explanations

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Makefile

@@ -2,7 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 BUILD_DIR = build
-SERVICES = users things http normalizer ws coap lora influxdb-writer influxdb-reader mongodb-writer mongodb-reader cassandra-writer cassandra-reader postgres-writer postgres-reader cli bootstrap opcua
+SERVICES = users things http ws coap lora normalizer influxdb-writer influxdb-reader mongodb-writer mongodb-reader cassandra-writer cassandra-reader postgres-writer postgres-reader cli bootstrap opcua
 DOCKERS = $(addprefix docker_,$(SERVICES))
 DOCKERS_DEV = $(addprefix docker_dev_,$(SERVICES))
 CGO_ENABLED ?= 0

cmd/cassandra-writer/main.go

@@ -19,6 +19,7 @@ import (
 	"github.com/gocql/gocql"
 	"github.com/mainflux/mainflux"
 	"github.com/mainflux/mainflux/logger"
+	"github.com/mainflux/mainflux/normalizer"
 	"github.com/mainflux/mainflux/writers"
 	"github.com/mainflux/mainflux/writers/api"
 	"github.com/mainflux/mainflux/writers/cassandra"
@@ -74,7 +75,8 @@ func main() {
 	defer session.Close()
 
 	repo := newService(session, logger)
-	if err := writers.Start(nc, repo, svcName, cfg.channels, logger); err != nil {
+	norm := normalizer.New()
+	if err := writers.Start(nc, repo, norm, svcName, cfg.channels, logger); err != nil {
 		logger.Error(fmt.Sprintf("Failed to create Cassandra writer: %s", err))
 	}
 

cmd/influxdb-writer/main.go

@@ -10,15 +10,14 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
-	"strconv"
 	"syscall"
-	"time"
 
 	"github.com/BurntSushi/toml"
 	kitprometheus "github.com/go-kit/kit/metrics/prometheus"
 	influxdata "github.com/influxdata/influxdb/client/v2"
 	"github.com/mainflux/mainflux"
 	"github.com/mainflux/mainflux/logger"
+	"github.com/mainflux/mainflux/normalizer"
 	"github.com/mainflux/mainflux/writers"
 	"github.com/mainflux/mainflux/writers/api"
 	"github.com/mainflux/mainflux/writers/influxdb"
@@ -29,43 +28,37 @@ import (
 const (
 	svcName = "influxdb-writer"
 
-	defNatsURL      = nats.DefaultURL
-	defLogLevel     = "error"
-	defPort         = "8180"
-	defBatchSize    = "5000"
-	defBatchTimeout = "5"
-	defDBName       = "mainflux"
-	defDBHost       = "localhost"
-	defDBPort       = "8086"
-	defDBUser       = "mainflux"
-	defDBPass       = "mainflux"
-	defChanCfgPath  = "/config/channels.toml"
-
-	envNatsURL      = "MF_NATS_URL"
-	envLogLevel     = "MF_INFLUX_WRITER_LOG_LEVEL"
-	envPort         = "MF_INFLUX_WRITER_PORT"
-	envBatchSize    = "MF_INFLUX_WRITER_BATCH_SIZE"
-	envBatchTimeout = "MF_INFLUX_WRITER_BATCH_TIMEOUT"
-	envDBName       = "MF_INFLUX_WRITER_DB_NAME"
-	envDBHost       = "MF_INFLUX_WRITER_DB_HOST"
-	envDBPort       = "MF_INFLUX_WRITER_DB_PORT"
-	envDBUser       = "MF_INFLUX_WRITER_DB_USER"
-	envDBPass       = "MF_INFLUX_WRITER_DB_PASS"
-	envChanCfgPath  = "MF_INFLUX_WRITER_CHANNELS_CONFIG"
+	defNatsURL     = nats.DefaultURL
+	defLogLevel    = "error"
+	defPort        = "8180"
+	defDBName      = "mainflux"
+	defDBHost      = "localhost"
+	defDBPort      = "8086"
+	defDBUser      = "mainflux"
+	defDBPass      = "mainflux"
+	defChanCfgPath = "/config/channels.toml"
+
+	envNatsURL     = "MF_NATS_URL"
+	envLogLevel    = "MF_INFLUX_WRITER_LOG_LEVEL"
+	envPort        = "MF_INFLUX_WRITER_PORT"
+	envDBName      = "MF_INFLUX_WRITER_DB_NAME"
+	envDBHost      = "MF_INFLUX_WRITER_DB_HOST"
+	envDBPort      = "MF_INFLUX_WRITER_DB_PORT"
+	envDBUser      = "MF_INFLUX_WRITER_DB_USER"
+	envDBPass      = "MF_INFLUX_WRITER_DB_PASS"
+	envChanCfgPath = "MF_INFLUX_WRITER_CHANNELS_CONFIG"
 )
 
 type config struct {
-	natsURL      string
-	logLevel     string
-	port         string
-	batchSize    string
-	batchTimeout string
-	dbName       string
-	dbHost       string
-	dbPort       string
-	dbUser       string
-	dbPass       string
-	channels     map[string]bool
+	natsURL  string
+	logLevel string
+	port     string
+	dbName   string
+	dbHost   string
+	dbPort   string
+	dbUser   string
+	dbPass   string
+	channels map[string]bool
 }
 
 func main() {
@@ -90,29 +83,13 @@ func main() {
 	}
 	defer client.Close()
 
-	batchTimeout, err := strconv.Atoi(cfg.batchTimeout)
-	if err != nil {
-		logger.Error(fmt.Sprintf("Invalid value for batch timeout: %s", err))
-		os.Exit(1)
-	}
-
-	batchSize, err := strconv.Atoi(cfg.batchSize)
-	if err != nil {
-		logger.Error(fmt.Sprintf("Invalid value of batch size: %s", err))
-		os.Exit(1)
-	}
-
-	timeout := time.Duration(batchTimeout) * time.Second
-	repo, err := influxdb.New(client, cfg.dbName, batchSize, timeout)
-	if err != nil {
-		logger.Error(fmt.Sprintf("Failed to create InfluxDB writer: %s", err))
-		os.Exit(1)
-	}
+	repo := influxdb.New(client, cfg.dbName)
 
 	counter, latency := makeMetrics()
 	repo = api.LoggingMiddleware(repo, logger)
 	repo = api.MetricsMiddleware(repo, counter, latency)
-	if err := writers.Start(nc, repo, svcName, cfg.channels, logger); err != nil {
+	norm := normalizer.New()
+	if err := writers.Start(nc, repo, norm, svcName, cfg.channels, logger); err != nil {
 		logger.Error(fmt.Sprintf("Failed to start InfluxDB writer: %s", err))
 		os.Exit(1)
 	}
@@ -133,17 +110,15 @@ func main() {
 func loadConfigs() (config, influxdata.HTTPConfig) {
 	chanCfgPath := mainflux.Env(envChanCfgPath, defChanCfgPath)
 	cfg := config{
-		natsURL:      mainflux.Env(envNatsURL, defNatsURL),
-		logLevel:     mainflux.Env(envLogLevel, defLogLevel),
-		port:         mainflux.Env(envPort, defPort),
-		batchSize:    mainflux.Env(envBatchSize, defBatchSize),
-		batchTimeout: mainflux.Env(envBatchTimeout, defBatchTimeout),
-		dbName:       mainflux.Env(envDBName, defDBName),
-		dbHost:       mainflux.Env(envDBHost, defDBHost),
-		dbPort:       mainflux.Env(envDBPort, defDBPort),
-		dbUser:       mainflux.Env(envDBUser, defDBUser),
-		dbPass:       mainflux.Env(envDBPass, defDBPass),
-		channels:     loadChansConfig(chanCfgPath),
+		natsURL:  mainflux.Env(envNatsURL, defNatsURL),
+		logLevel: mainflux.Env(envLogLevel, defLogLevel),
+		port:     mainflux.Env(envPort, defPort),
+		dbName:   mainflux.Env(envDBName, defDBName),
+		dbHost:   mainflux.Env(envDBHost, defDBHost),
+		dbPort:   mainflux.Env(envDBPort, defDBPort),
+		dbUser:   mainflux.Env(envDBUser, defDBUser),
+		dbPass:   mainflux.Env(envDBPass, defDBPass),
+		channels: loadChansConfig(chanCfgPath),
 	}
 
 	clientCfg := influxdata.HTTPConfig{

cmd/mongodb-writer/main.go

@@ -17,6 +17,7 @@ import (
 	kitprometheus "github.com/go-kit/kit/metrics/prometheus"
 	"github.com/mainflux/mainflux"
 	"github.com/mainflux/mainflux/logger"
+	"github.com/mainflux/mainflux/normalizer"
 	"github.com/mainflux/mainflux/writers"
 	"github.com/mainflux/mainflux/writers/api"
 	"github.com/mainflux/mainflux/writers/mongodb"
@@ -84,7 +85,8 @@ func main() {
 	counter, latency := makeMetrics()
 	repo = api.LoggingMiddleware(repo, logger)
 	repo = api.MetricsMiddleware(repo, counter, latency)
-	if err := writers.Start(nc, repo, svcName, cfg.channels, logger); err != nil {
+	norm := normalizer.New()
+	if err := writers.Start(nc, repo, norm, svcName, cfg.channels, logger); err != nil {
 		logger.Error(fmt.Sprintf("Failed to start MongoDB writer: %s", err))
 		os.Exit(1)
 	}

cmd/postgres-writer/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/jmoiron/sqlx"
 	"github.com/mainflux/mainflux"
 	"github.com/mainflux/mainflux/logger"
+	"github.com/mainflux/mainflux/normalizer"
 	"github.com/mainflux/mainflux/writers"
 	"github.com/mainflux/mainflux/writers/api"
 	"github.com/mainflux/mainflux/writers/postgres"
@@ -80,7 +81,8 @@ func main() {
 	defer db.Close()
 
 	repo := newService(db, logger)
-	if err = writers.Start(nc, repo, svcName, cfg.channels, logger); err != nil {
+	norm := normalizer.New()
+	if err = writers.Start(nc, repo, norm, svcName, cfg.channels, logger); err != nil {
 		logger.Error(fmt.Sprintf("Failed to create Postgres writer: %s", err))
 	}
 

docker/addons/bootstrap/docker-compose.yml

@@ -1,3 +1,11 @@
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
+# This docker-compose file contains optional bootstrap services. Since it's optional, this file is
+# dependent of docker-compose file from <project_root>/docker. In order to run this services, execute command:
+# docker-compose -f docker/docker-compose.yml -f docker/addons/bootstrap/docker-compose.yml up
+# from project root.
+
 version: "3.7"
 
 networks:

docker/addons/cassandra-reader/docker-compose.yml

@@ -1,10 +1,10 @@
-###
-# This docker-compose file contains optional Cassandra and cassandra-reader. Since these are optional, this file is
-# dependent of docker-compose file from <project_root>/docker. In order to run
-# these optional service, execute command:
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
+# This docker-compose file contains optional cassandra-reader. Since it's optional, this file is
+# dependent of docker-compose file from <project_root>/docker. In order to run this service, execute command:
 # docker-compose -f docker/docker-compose.yml -f docker/addons/cassandra-reader/docker-compose.yml up
 # from project root.
-###
 
 version: "3.7"
 

docker/addons/cassandra-writer/docker-compose.yml

@@ -1,10 +1,10 @@
-###
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
 # This docker-compose file contains optional Cassandra and cassandra-writer. Since these are optional, this file is
-# dependent of docker-compose file from <project_root>/docker. In order to run
-# these optional service, execute command:
+# dependent of docker-compose file from <project_root>/docker. In order to run these services, execute command:
 # docker-compose -f docker/docker-compose.yml -f docker/addons/cassandra-writer/docker-compose.yml up
 # from project root.
-###
 
 version: "3.7"
 

docker/addons/influxdb-reader/docker-compose.yml

@@ -1,7 +1,10 @@
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
 ###
 # This docker-compose file contains optional InfluxDB-reader service for the Mainflux
 # platform. Since this service is optional, this file is dependent on the docker-compose.yml
-# file from <project_root>/docker/. In order to run InfluxDB-reader service, core services,
+# file from <project_root>/docker/. In order to run this service, core services,
 # as well as the network from the core composition, should be already running.
 ###
 

docker/addons/influxdb-writer/docker-compose.yml

@@ -1,9 +1,10 @@
-###
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
 # This docker-compose file contains optional InfluxDB, InfluxDB-writer and Grafana services
 # for the Mainflux platform. Since this services are optional, this file is dependent on the
 # docker-compose.yml file from <project_root>/docker/. In order to run these services,
 # core services, as well as the network from the core composition, should be already running.
-###
 
 version: "3.7"
 
@@ -16,7 +17,6 @@ volumes:
   mainflux-grafana-volume:
 
 services:
-
   influxdb:
     image: influxdb:1.6.4-alpine
     container_name: mainflux-influxdb

docker/addons/lora-adapter/docker-compose.yml

@@ -1,9 +1,10 @@
-###
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
 # This docker-compose file contains optional lora-adapter and lora-redis services
 # for the Mainflux platform. Since this services are optional, this file is dependent on the
 # docker-compose.yml file from <project_root>/docker/. In order to run these services,
 # core services, as well as the network from the core composition, should be already running.
-###
 
 version: "3.7"
 

docker/addons/mongodb-reader/docker-compose.yml

@@ -1,10 +1,11 @@
-###
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
 # This docker-compose file contains optional MongoDB-reader service
 # for Mainflux platform. Since these are optional, this file is dependent of docker-compose file
-# from <project_root>/docker. In order to run these optional service, execute command:
+# from <project_root>/docker. In order to run this service, execute command:
 # docker-compose -f docker/docker-compose.yml -f docker/addons/mongodb-reader/docker-compose.yml up
 # from project root. MongoDB service is defined in docker/addons/mongodb-writer/docker-compose.yml.
-###
 
 version: "3.7"
 

docker/addons/mongodb-writer/docker-compose.yml

@@ -1,11 +1,12 @@
-###
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
 # This docker-compose file contains optional MongoDB and MongoDB-writer services
 # for Mainflux platform. Since these are optional, this file is dependent of docker-compose file
-# from <project_root>/docker. In order to run these optional service, execute command:
+# from <project_root>/docker. In order to run these services, execute command:
 # docker-compose -f docker/docker-compose.yml -f docker/addons/mongodb-writer/docker-compose.yml up
 # from project root. MongoDB default port (27017) is exposed, so you can use various tools for database
 # inspection and data visualization.
-###
 
 version: "3.7"
 

docker/addons/normalizer/docker-compose.yml

@@ -0,0 +1,29 @@
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
+# This docker-compose file contains optional lora-adapter and lora-redis services
+# for the Mainflux platform. Since this service is optional, this file is dependent on the
+# docker-compose.yml file from <project_root>/docker/. In order to run this service,
+# core services, as well as the network from the core composition, should be already running.
+
+version: "3.7"
+
+networks:
+  docker_mainflux-base-net:
+    external: true
+
+services:
+  normalizer:
+    image: mainflux/normalizer:latest
+    container_name: mainflux-normalizer
+    restart: on-failure
+    depends_on:
+      - nats
+    environment:
+      MF_NORMALIZER_LOG_LEVEL: ${MF_NORMALIZER_LOG_LEVEL}
+      MF_NATS_URL: ${MF_NATS_URL}
+      MF_NORMALIZER_PORT: ${MF_NORMALIZER_PORT}
+    ports:
+      - ${MF_NORMALIZER_PORT}:${MF_NORMALIZER_PORT}
+    networks:
+      - mainflux-base-net

docker/addons/opcua-adapter/docker-compose.yml

@@ -1,9 +1,10 @@
-###
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
 # This docker-compose file contains optional opcua-adapter and opcua-redis services
 # for the Mainflux platform. Since this services are optional, this file is dependent on the
 # docker-compose.yml file from <project_root>/docker/. In order to run these services,
 # core services, as well as the network from the core composition, should be already running.
-###
 
 version: "3.7"
 

docker/addons/postgres-reader/docker-compose.yml

@@ -1,10 +1,11 @@
-###
+# Copyright (c) Mainflux
+# SPDX-License-Identifier: Apache-2.0
+
 # This docker-compose file contains optional Postgres-reader service for Mainflux platform.
 # Since this service is optional, this file is dependent of docker-compose.yml file
-# from <project_root>/docker. In order to run these optional service, execute command:
+# from <project_root>/docker. In order to run this service, execute command:
 # docker-compose -f docker/docker-compose.yml -f docker/addons/postgres-reader/docker-compose.yml up
 # from project root.
-###
 
 version: "3.7"