Merge pull request #69 from HyperloopUPV-H8/backend/logger-implementation-hotfix

[backend] logger-implementation

Author: msanlli

.gitignore

@@ -1,4 +1,5 @@
 build
+.vscode
 
 # GOOGLE API KEY
 secret.json

backend/cmd/main.go

@@ -22,17 +22,11 @@ import (
 	"github.com/HyperloopUPV-H8/h9-backend/internal/excel/ade"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/excel/utils"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/info"
-	"github.com/HyperloopUPV-H8/h9-backend/internal/logger_handler"
-	protection_logger "github.com/HyperloopUPV-H8/h9-backend/internal/message_logger"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/message_transfer"
-	"github.com/HyperloopUPV-H8/h9-backend/internal/order_logger"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/order_transfer"
-	"github.com/HyperloopUPV-H8/h9-backend/internal/packet_logger"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/pod_data"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/server"
-	"github.com/HyperloopUPV-H8/h9-backend/internal/state_space_logger"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/update_factory"
-	"github.com/HyperloopUPV-H8/h9-backend/internal/value_logger"
 	vehicle_models "github.com/HyperloopUPV-H8/h9-backend/internal/vehicle/models"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/ws_handle"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
@@ -51,6 +45,12 @@ import (
 	"github.com/google/gopacket/pcap"
 	"github.com/pelletier/go-toml/v2"
 	trace "github.com/rs/zerolog/log"
+
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/logger"
+	data_logger "github.com/HyperloopUPV-H8/h9-backend/pkg/logger/data"
+	messages_logger "github.com/HyperloopUPV-H8/h9-backend/pkg/logger/messages"
+	order_logger "github.com/HyperloopUPV-H8/h9-backend/pkg/logger/order"
+	state_logger "github.com/HyperloopUPV-H8/h9-backend/pkg/logger/state"
 )
 
 var traceLevel = flag.String("trace", "info", "set the trace level (\"fatal\", \"error\", \"warn\", \"info\", \"debug\", \"trace\")")
@@ -124,20 +124,15 @@ func main() {
 	updateFactory := update_factory.NewFactory()
 
 	// <--- logger --->
-	packetLogger := packet_logger.NewPacketLogger(podData.Boards, config.PacketLogger)
-	valueLogger := value_logger.NewValueLogger(podData.Boards, config.ValueLogger)
-	orderLogger := order_logger.NewOrderLogger(podData.Boards, config.OrderLogger)
-	protectionLogger := protection_logger.NewMessageLogger(config.Vehicle.Messages.InfoIdKey, config.Vehicle.Messages.FaultIdKey, config.Vehicle.Messages.WarningIdKey, config.ProtectionLogger)
-	stateSpaceLogger := state_space_logger.NewStateSpaceLogger(info.MessageIds.StateSpace)
-
-	loggers := map[string]logger_handler.Logger{
-		"packets":     &packetLogger,
-		"values":      &valueLogger,
-		"orders":      &orderLogger,
-		"protections": &protectionLogger,
-		"stateSpace":  &stateSpaceLogger,
+	var boardMap map[abstraction.BoardId]string
+	var subloggers = map[abstraction.LoggerName]abstraction.Logger{
+		data_logger.Name:     data_logger.NewLogger(),
+		messages_logger.Name: messages_logger.NewLogger(boardMap),
+		order_logger.Name:    order_logger.NewLogger(),
+		state_logger.Name:    state_logger.NewLogger(),
 	}
-	loggerHandler := logger_handler.NewLoggerHandler(loggers, config.LoggerHandler)
+
+	loggerHandler := logger.NewLogger(subloggers)
 
 	// <--- order transfer --->
 	idToBoard := make(map[uint16]string)
@@ -167,37 +162,64 @@ func main() {
 	transp.SetAPI(&TransportAPI{
 		OnNotification: func(notification abstraction.TransportNotification) {
 			packet := notification.(transport.PacketNotification)
+
 			switch p := packet.Packet.(type) {
 			case *data.Packet:
-				if _, ok := orders[p.Id()]; ok {
-					loggerHandler.Log(order_logger.LoggableOrder(*p))
-					return
-				}
-
 				update := updateFactory.NewUpdate(p)
 				dataTransfer.Update(update)
 
-				loggerHandler.Log(packet_logger.ToLoggablePacket(p))
+				err = loggerHandler.PushRecord(&data_logger.Record{
+					Packet: p,
+				})
 
-				for id, value := range p.GetValues() {
-					loggerHandler.Log(value_logger.ToLoggableValue(string(id), value, p.Timestamp()))
+				if err != nil {
+					fmt.Println("Error pushing record to logger: ", err)
 				}
+
 			case *info_packet.Packet:
 				messageTransfer.SendMessage(p)
-				loggerHandler.Log(protection_logger.LoggableInfo(*p))
+
+				err = loggerHandler.PushRecord(&messages_logger.Record{
+					Packet: p,
+				})
+
+				if err != nil {
+					fmt.Println("Error pushing record to logger: ", err)
+				}
+
 			case *protection.Packet:
 				messageTransfer.SendMessage(p)
-				loggerHandler.Log(protection_logger.LoggableProtection(*p))
+
+				packet := info_packet.NewPacket(p.Id())
+				packet.BoardId = p.BoardId
+				packet.Timestamp = p.Timestamp
+				packet.Msg = info_packet.InfoData(fmt.Sprint(p))
+
+				err = loggerHandler.PushRecord(&messages_logger.Record{
+					Packet: packet,
+				})
+
+				if err != nil {
+					fmt.Println("Error pushing record to logger: ", err)
+				}
+
 			case *blcu_packet.Ack:
 				if useBlcu {
 					blcu.NotifyAck()
 				}
+
 			case *state.Space:
-				for _, row := range p.State() {
-					loggerHandler.Log(state_space_logger.LoggableStateSpaceRow(row))
+				err = loggerHandler.PushRecord(&state_logger.Record{
+					Packet: p,
+				})
+
+				if err != nil {
+					fmt.Println("Error pushing record to logger: ", err)
 				}
+
 			case *order.Add:
 				orderTransfer.AddStateOrders(*p)
+
 			case *order.Remove:
 				orderTransfer.RemoveStateOrders(*p)
 			}
@@ -259,7 +281,13 @@ func main() {
 				trace.Error().Any("order", order).Err(err).Msg("error sending order")
 			}
 
-			loggerHandler.Log(order_logger.LoggableOrder(order))
+			err = loggerHandler.PushRecord(&order_logger.Record{
+				Packet: &order,
+			})
+
+			if err != nil {
+				fmt.Println("Error pushing record to logger: ", err)
+			}
 		}
 	}()
 
@@ -285,7 +313,7 @@ func main() {
 
 	websocketBroker.RegisterHandle(&connectionTransfer, config.Connections.UpdateTopic, "connection/update")
 	websocketBroker.RegisterHandle(&dataTransfer, "podData/update")
-	websocketBroker.RegisterHandle(&loggerHandler, config.LoggerHandler.Topics.Enable)
+	websocketBroker.RegisterHandle(loggerHandler, config.LoggerHandler.Topics.Enable)
 	websocketBroker.RegisterHandle(&messageTransfer, "message/update")
 	websocketBroker.RegisterHandle(&orderTransfer, config.Orders.SendTopic, "order/stateOrders")
 

backend/pkg/abstraction/logger.go

@@ -16,8 +16,8 @@ type LoggerRequest interface {
 // Logger is the module in charge of storing and retrieving information from the
 // hard drive, making it persistent.
 type Logger interface {
-	Start(startKeys []LoggerName) error
-	Stop(stopKeys []LoggerName) error
+	Start() error
+	Stop() error
 	// PushRecord will store a record to disk
 	PushRecord(LoggerRecord) error
 	// PullRecord will retrieve a record from disk

backend/pkg/logger/CHANGELOG.md

@@ -1,28 +1,28 @@
 # Logger Changelog
 
-## v0.0.0
+## v0
 
-### v0.3.0
+### v0.3
 
-#### v0.3.1
+#### v0.3.0
 
 - Integrate abstraction
 - "Keys" concept implementation
 - Data sublogger
   - Add constants
   - Add map of valueTypes
 
-#### v0.3.2
+#### v0.3.1
 
 - Errors and name publicity fixed
 
-#### v0.3.3
+#### v0.3.2
 
 - Change from map to list
 - Use more descriptive names
 - Fix constants
 
-#### v0.3.4
+#### v0.3.3
 
 - Change type check logic
 - Handle writer errors
@@ -31,19 +31,31 @@
 - Prevent concurrency-related errors
 - Logic fixes
 
-## v1.0.0
+## v1
 
-### v1.1.0
+### v1.0
 
-#### v1.1.1
+#### v1.0.0
 
 - Messages sublogger
 - Handle json timestamp
 
-#### v1.1.2
+#### v1.0.1
 
 - Order sublogger
 
-#### v1.1.3
+#### v1.0.2
 
 - State sublogger
+
+### v1.2
+
+#### v1.2.0
+
+- Implement logger into cmd
+- Fix keys logic
+- Fix log logic
+
+#### v1.2.1
+
+- Change logging paths

backend/pkg/logger/data/logger.go

@@ -25,27 +25,35 @@ type Logger struct {
 	// An atomic boolean is used in order to use CompareAndSwap in the Start and Stop methods
 	running  *atomic.Bool
 	fileLock *sync.RWMutex
-	// initialTime fixes the starting time of the log
-	initialTime time.Time
 	// valueFileSlice is a map that contains the file of each value
 	valueFileSlice map[data.ValueName]io.WriteCloser
 }
 
 // Record is a struct that implements the abstraction.LoggerRecord interface
 type Record struct {
-	packet *data.Packet
+	Packet *data.Packet
 }
 
-func (data *Record) Name() abstraction.LoggerName {
+func (record *Record) Name() abstraction.LoggerName {
 	return Name
 }
 
+func NewLogger() *Logger {
+	logger := &Logger{
+		valueFileSlice: make(map[data.ValueName]io.WriteCloser),
+		running:        &atomic.Bool{},
+		fileLock:       &sync.RWMutex{},
+	}
+
+	logger.running.Store(false)
+	return logger
+}
+
 func (sublogger *Logger) Start() error {
 	if !sublogger.running.CompareAndSwap(false, true) {
 		fmt.Println("Logger already running")
 		return nil
 	}
-	sublogger.initialTime = time.Now()
 
 	fmt.Println("Logger started")
 	return nil
@@ -74,15 +82,15 @@ func (sublogger *Logger) PushRecord(record abstraction.LoggerRecord) error {
 		}
 	}
 
-	valueMap := dataRecord.packet.GetValues()
+	valueMap := dataRecord.Packet.GetValues()
 
 	sublogger.fileLock.Lock()
 	defer sublogger.fileLock.Unlock()
 
 	writerErr := error(nil)
 	for valueName, value := range valueMap {
-		var packet *Record
-		timestamp := packet.packet.Timestamp()
+
+		timestamp := dataRecord.Packet.Timestamp()
 
 		var val string
 
@@ -99,7 +107,10 @@ func (sublogger *Logger) PushRecord(record abstraction.LoggerRecord) error {
 
 		file, ok := sublogger.valueFileSlice[valueName]
 		if !ok {
-			f, err := os.Create(path.Join(string(valueName), fmt.Sprintf("%s_%s.csv", valueName, packet.packet.Timestamp().Format("3339"))))
+			filename := path.Join("logger/data", fmt.Sprintf("data_%s", logger.Timestamp.Format(time.RFC3339)), fmt.Sprintf("%s.csv", valueName))
+			os.MkdirAll(path.Dir(filename), os.ModePerm)
+
+			f, err := os.Create(path.Join(filename))
 			if err != nil {
 				return &logger.ErrCreatingFile{
 					Name:      Name,
@@ -125,11 +136,16 @@ func (sublogger *Logger) PullRecord(request abstraction.LoggerRequest) (abstract
 	panic("TODO!")
 }
 
-func Stop(sublogger *Logger) {
+func (sublogger *Logger) Stop() error {
 	if !sublogger.running.CompareAndSwap(true, false) {
 		fmt.Println("Logger already stopped")
-		return
+		return nil
+	}
+
+	for _, file := range sublogger.valueFileSlice {
+		file.Close()
 	}
 
 	fmt.Println("Logger stopped")
+	return nil
 }

backend/pkg/logger/errors.go

@@ -48,3 +48,11 @@ type ErrWrongRecordType struct {
 func (err *ErrWrongRecordType) Error() string {
 	return fmt.Sprintf("Wrong record type for logger %s at %s, expected %T, got %T", err.Name, err.Timestamp.Format(time.RFC3339), err.Expected.Name(), err.Received.Name())
 }
+
+type ErrParsingLoggerMap struct {
+	Name abstraction.LoggerName
+}
+
+func (err ErrParsingLoggerMap) Error() string {
+	return fmt.Sprintf("Error parsing logger map for logger %s", err.Name)
+}