processor

func Clean ¶

func Clean(ctx context.Context, pipe <-chan *pipelines.LogEntry, wg *sync.WaitGroup, mu *sync.Mutex, Shipper pkgEx.Shipper)

Clean function that processes the received information and sends it to their corresponding functions

Parameters:

• ctx: context for cancellation (not explicitly used in the current body)
• pipe: input channel of *pipelines.LogEntry. Processing continues until the channel is closed and all remaining entries are drained.
• wg: WaitGroup to notify the caller that the goroutine has completed
• mu: mutex protecting the shared maps 'logs' and 'logsBurst'.

The function performs:

• Sanitization and clustering of log content.
• Statistics updates by source and message (logs).
• Burst detection by log level (logsBurst).
• Returning processed objects back to pools (EntryPool, BufferPool).
• Periodic exporting triggered by Ticker.C Sends logs to exporter.ShipLogs, exporter.Console and integrations.Integrations. If GenerateLogsFile is enabled, accumulates logs for later writing.
• File Export triggered by TickerReportFile.C when GenerateLogsFile is true.

Note: This functions is intented to run as a concurrent goroutine. It uses the unsafe package for zero-copy byte-to-string conversions, assuming the underlying buffers will not be modified afterward.

func CleanFile ¶ added in v1.4.0

func CleanFile(file *os.File, idLog string, mu *sync.Mutex, routeFile string, Shipper pkgEx.Shipper)

CleanFile processes a log file line by line, applying sanitization and clustering, accumulating statistics, and sending them to exporters (ShipLogs and FileReader).

Parameters:

• file: pointer to the opened file to be read.
• idLog: log identifier (passed to cluster.Cluster).
• mu: mutex (maintained for consistency with other signatures, though logsFile is local).
• routeFile: path of the original file (passed to exporter.FileReader for reference).

The function performs:

• Initializes a local map 'logsFile' to accumulate statistics under the "file-reader" source.

• Creates a bufio.Scanner to read the file line by line.

• For each line:

  -Reads line bytes.

  -Calls cluster.Cluster(data, idLog) to obtain sanitized and grouped text.

  -Converts the result to a string using 'unsafe' (zero-copy) and assigns it to sanitizedText.

  -Acquires the mutex (kept for structural consistency).

  -Initializes the entry in logsFile["file-reader"] if necessary.

  -Extracts the log level using sanitizer.ExtractLevelUpper(data) and converts it to string via 'unsafe'.

  -If the sanitized message is new, creates an entry in logsFile with initial statistics.

  -Increments the counter and updates LastSeen.

  -Releases the mutex.

• After reaching EOF, calls exporter.ShipLogs(logsFile) to send logs to the remote shipper.

• Calls exporter.FileReader(logsFile, routeFile) to save the report to a file.

• Calls integrations.Integrations(logsFile), send data to Observability Platforms

• Checks for scanner errors via scanner.Err() and terminates with log.Fatal if found.

• Clears the logsFile map using 'clear()' to free memory.