Simo dev

.gitignore

@@ -15,3 +15,4 @@ grafana
 trackeroo-backend/mongo/data
 nominatim-data
 osrm-data
+mqtt_client

trackeroo-backend/brokeroo/cmd/brokeroo/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"fmt"
@@ -13,6 +14,7 @@ import (
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	_ "github.com/lib/pq"
+	"github.com/segmentio/kafka-go"
 )
 
 type Config struct {
@@ -22,6 +24,7 @@ type Config struct {
 	MQTTPassword string
 	PostgresURL  string
 	TopicPattern string
+	KafkaBroker  string
 }
 
 type MQTTData struct {
@@ -32,14 +35,17 @@ type MQTTData struct {
 }
 
 type Service struct {
-	config     *Config
-	mqttClient mqtt.Client
-	db         *sql.DB
+	config      *Config
+	mqttClient  mqtt.Client
+	db          *sql.DB
+	kafkaWriter *kafka.Writer
+	knownTopics map[string]bool /* for caching kafka topics  */
 }
 
 func NewService(config *Config) *Service {
 	return &Service{
-		config: config,
+		config:      config,
+		knownTopics: make(map[string]bool),
 	}
 }
 
@@ -57,6 +63,68 @@ func (s *Service) connectPostgres() error {
 	return nil
 }
 
+func (s *Service) connectKafka() error {
+	s.kafkaWriter = kafka.NewWriter(kafka.WriterConfig{
+		Brokers: []string{s.config.KafkaBroker},
+	})
+
+	if err := s.ensureTopicExists("health-check"); err != nil {
+		log.Printf("failed to create first topic for healt-check: %v", err)
+	}
+
+	err := s.kafkaWriter.WriteMessages(context.Background(),
+		kafka.Message{
+			Topic: "health-check",
+			Value: []byte("ping"),
+		},
+	)
+	if err != nil {
+		log.Printf("Failed to write test message to Kafka: %v", err)
+		return err
+	}
+
+	log.Println("Successfully connected to Kafka at", s.config.KafkaBroker)
+	return nil
+}
+
+func (s *Service) ensureTopicExists(topic string) error {
+
+	if s.knownTopics[topic] {
+		return nil
+	}
+
+	conn, err := kafka.Dial("tcp", s.config.KafkaBroker)
+	if err != nil {
+		return fmt.Errorf("dial broker: %w", err)
+	}
+	defer conn.Close()
+
+	controller, err := conn.Controller()
+	if err != nil {
+		return fmt.Errorf("get controller: %w", err)
+	}
+
+	controllerConn, err := kafka.Dial("tcp", fmt.Sprintf("%s:%d", controller.Host, controller.Port))
+	if err != nil {
+		return fmt.Errorf("dial controller: %w", err)
+	}
+	defer controllerConn.Close()
+
+	err = controllerConn.CreateTopics(kafka.TopicConfig{
+		Topic:             topic,
+		NumPartitions:     1,
+		ReplicationFactor: 1,
+	})
+
+	if err != nil && !strings.Contains(err.Error(), "Topic with this name already exists") {
+		return fmt.Errorf("create topic %s: %w", topic, err)
+	}
+
+	s.knownTopics[topic] = true
+	log.Printf("Topic pronto: %s", topic)
+	return nil
+}
+
 func (s *Service) connectMQTT() error {
 	opts := mqtt.NewClientOptions()
 	opts.AddBroker(s.config.MQTTBroker)
@@ -124,19 +192,51 @@ func (s *Service) messageHandler(client mqtt.Client, msg mqtt.Message) {
 
 	// Get current Unix timestamp
 	tsUnix := time.Now().Unix()
+	ts := time.Now()
 
-	if err := s.insertData(tsUnix, devID, tag, jsonPayload); err != nil {
+	/* postgress */
+	if err := s.insertData(ts, tsUnix, devID, tag, jsonPayload); err != nil {
 		log.Printf("Failed to insert data: %v", err)
 		return
 	}
 
+	/* kafka */
+	kafkaTopic := sanitizeTopic(topic)
+
+	/*TODO check ctx for timeout
+	tx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	err := s.kafkaWriter.WriteMessages(ctx, kafka.Message{
+	    Topic: kafkaTopic,
+	    Key:   []byte(devID),
+	    Value: payload,
+	})
+	*/
+	if err := s.ensureTopicExists(kafkaTopic); err != nil {
+		log.Printf("Errore creazione topic %s: %v", kafkaTopic, err)
+		return
+	}
+
+	err := s.kafkaWriter.WriteMessages(context.Background(),
+		kafka.Message{
+			Topic: kafkaTopic,
+			Value: payload,
+		},
+	)
+	if err != nil {
+		log.Printf("Failed to write to Kafka: %v", err)
+	} else {
+		log.Printf("Message forwarded to Kafka topic: %s", kafkaTopic)
+	}
+
 	log.Printf("Successfully inserted data for dev_id: %s, tag: %s", devID, tag)
 }
 
-func (s *Service) insertData(tsUnix int64, devID, tag string, payload json.RawMessage) error {
-	query := `INSERT INTO trackeroo.data (ts_unix, dev_id, tag, payload) VALUES ($1, $2, $3, $4)`
+func (s *Service) insertData(ts time.Time, tsUnix int64, devID, tag string, payload json.RawMessage) error {
+	query := `INSERT INTO trackeroo.data (ts, ts_unix, dev_id, tag, payload) VALUES ($1, $2, $3, $4, $5)`
 
-	_, err := s.db.Exec(query, tsUnix, devID, tag, payload)
+	_, err := s.db.Exec(query, ts, tsUnix, devID, tag, payload)
 	if err != nil {
 		return fmt.Errorf("failed to insert into database: %w", err)
 	}
@@ -155,6 +255,10 @@ func (s *Service) Start() error {
 		return err
 	}
 
+	// Connect to Kafka
+	if err := s.connectKafka(); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -171,6 +275,11 @@ func (s *Service) Stop() {
 		s.db.Close()
 		log.Println("Closed PostgreSQL connection")
 	}
+
+	if s.kafkaWriter != nil {
+		s.kafkaWriter.Close()
+		log.Println("Closed Kafka writer")
+	}
 }
 
 func loadConfig() *Config {
@@ -181,6 +290,7 @@ func loadConfig() *Config {
 		MQTTPassword: getEnvOrDefault("MQTT_PASSWORD", ""),
 		PostgresURL:  getEnvOrDefault("POSTGRES_URL", "postgres://user:password@localhost/dbname?sslmode=disable"),
 		TopicPattern: getEnvOrDefault("TOPIC_PATTERN", "j/data/+/+"),
+		KafkaBroker:  getEnvOrDefault("KAFKA_BROKER", "kafka:9092"),
 	}
 }
 
@@ -191,6 +301,10 @@ func getEnvOrDefault(key, defaultValue string) string {
 	return defaultValue
 }
 
+func sanitizeTopic(topic string) string {
+	return strings.ReplaceAll(topic, "/", "-")
+}
+
 func main() {
 	log.Println("Starting MQTT to PostgreSQL service...")
 

trackeroo-backend/brokeroo/go.mod

@@ -7,8 +7,14 @@ require (
 	github.com/lib/pq v1.10.9
 )
 
+require (
+	github.com/klauspost/compress v1.15.9 // indirect
+	github.com/pierrec/lz4/v4 v4.1.15 // indirect
+)
+
 require (
 	github.com/gorilla/websocket v1.5.3 // indirect
-	golang.org/x/net v0.27.0 // indirect
+	github.com/segmentio/kafka-go v0.4.49
+	golang.org/x/net v0.38.0 // indirect
 	golang.org/x/sync v0.7.0 // indirect
 )

trackeroo-backend/brokeroo/go.sum

@@ -2,9 +2,17 @@ github.com/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjO
 github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
+github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0=
+github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/segmentio/kafka-go v0.4.49 h1:GJiNX1d/g+kG6ljyJEoi9++PUMdXGAxb7JGPiDCuNmk=
+github.com/segmentio/kafka-go v0.4.49/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E=
 golang.org/x/net v0.27.0 h1:5K3Njcw06/l2y9vpGCSdcxWOYHOUk3dVNGDXN+FvAys=
 golang.org/x/net v0.27.0/go.mod h1:dDi0PyhWNoiUOrAS8uXv/vnScO4wnHQO4mj9fn/RytE=
+golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8=
+golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
 golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
 golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=

trackeroo-backend/docker-compose.yml

@@ -22,6 +22,8 @@ services:
       MQTT_HOST: localhost
       MQTT_PORT: 1883
       MQTTS_PORT: 8883
+    volumes:
+      - trackeroo-backend-data:/app/data
 
   mongo:
     image: mongo:latest
@@ -33,27 +35,24 @@ services:
       retries: 5
       start_period: 5s
     volumes:
+      - mongo-data:/data/db
       - ./mongo/init.js:/docker-entrypoint-initdb.d/init.js:ro
-      - ./mongo/data:/data/db
 
   rabbitmq:
     image: rabbitmq:3-management
     container_name: rabbitmq
     ports:
-      - "5672:5672" # AMQP
-      - "15672:15672" # Management UI
-      - "1883:1883" # MQTT
-      - "8883:8883" # MQTT
+      - "5672:5672"
+      - "15672:15672"
+      - "1883:1883"
+      - "8883:8883"
     environment:
       RABBITMQ_DEFAULT_USER: admin
       RABBITMQ_DEFAULT_PASS: admin123
       RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: '-rabbitmq_mqtt tcp_listeners [{"0.0.0.0",1883}]'
     volumes:
+      - rabbitmq-data:/var/lib/rabbitmq
       - ./rabbitmq/etc:/etc/rabbitmq/
-      - ./rabbitmq/data:/var/lib/rabbitmq
-    command: >
-      bash -c "rabbitmq-plugins enable --offline rabbitmq_mqtt;
-              rabbitmq-server"
 
   tsdb:
     image: ghcr.io/skiby7/tsdb:latest
@@ -63,16 +62,14 @@ services:
       timeout: 5s
       retries: 5
     container_name: tsdb
+    command: ["postgres", "-c", "shared_preload_libraries=timescaledb"]
     environment:
       POSTGRES_DB: tracker_db
       POSTGRES_USER: postgres
       POSTGRES_PASSWORD: postgres
-
-      # Custom Postgres parameters
       POSTGRES_INITDB_ARGS: "--auth-host=scram-sha-256 --auth-local=scram-sha-256"
-
     volumes:
-      - ./tsdb/data:/var/lib/postgresql/data
+      - tsdb-data:/var/lib/postgresql/data
       - ./tsdb/initdb:/docker-entrypoint-initdb.d
 
   pgadmin:
@@ -85,7 +82,7 @@ services:
     ports:
       - "8085:80"
     volumes:
-      - ./tsdb/pgadmin_data:/var/lib/pgadmin
+      - pgadmin-data:/var/lib/pgadmin
     depends_on:
       tsdb:
         condition: service_healthy
@@ -102,10 +99,13 @@ services:
       POSTGRES_URL: postgres://admin:administrator@tsdb:5432/tracker_db?sslmode=disable
       TOPIC_PATTERN: j/data/+/+
     depends_on:
+      kafka: 
+        condition: service_started
       tsdb:
         condition: service_healthy
       rabbitmq:
         condition: service_started
+
   grafana:
     image: grafana/grafana:latest
     container_name: grafana
@@ -114,11 +114,8 @@ services:
       tsdb:
         condition: service_healthy
     environment:
-      # Default admin user
       GF_SECURITY_ADMIN_USER: admin
       GF_SECURITY_ADMIN_PASSWORD: admin123
-
-      # Optional: Configure Grafana to use Postgres as its own database
       GF_DATABASE_TYPE: postgres
       GF_DATABASE_HOST: tsdb:5432
       GF_DATABASE_NAME: tracker_db
@@ -127,7 +124,35 @@ services:
     ports:
       - "3000:3000"
     volumes:
-      - ./grafana:/var/lib/grafana
+      - grafana-data:/var/lib/grafana
+
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.5.0
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  kafka:
+    image: confluentinc/cp-kafka:7.5.0
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT 
+    depends_on:
+      - zookeeper
+
+volumes:
+  trackeroo-backend-data:
+  mongo-data:
+  rabbitmq-data:
+  tsdb-data:
+  pgadmin-data:
+  grafana-data:
+
 secrets:
   users_key:
-    file: secrets/users_key
+    file: secrets/users_key