watch-tool/service_monitor.go

package main

import (
	"bufio"
	"codeberg.org/pata1704/drain3"
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"os/exec"
	"strconv"
	"strings"
	"time"
	"watch-tool/models"
	"watch-tool/parser"
)

type ServiceMonitor struct {
	config      ServiceConfig
	hostname    string
	drainConfig *drain3.Config
	stateDir    string
}

func NewServiceMonitor(config ServiceConfig, hostname string, drainCfg *drain3.Config, stateDir string) *ServiceMonitor {
	return &ServiceMonitor{
		config:      config,
		hostname:    hostname,
		drainConfig: drainCfg,
		stateDir:    stateDir,
	}
}

func (sm *ServiceMonitor) Start(ctx context.Context, out chan<- models.LogMessage) error {
	args := sm.buildJournalctlArgs()

	slog.Info("starting journalctl", "arguments", args)

	cmd := exec.CommandContext(ctx, args[0], args[1:]...)

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return fmt.Errorf("error StdoutPipe: %w", err)
	}

	if err := cmd.Start(); err != nil {
		return fmt.Errorf("error start command: %w", err)
	}

	scanner := bufio.NewScanner(stdout)

	go func() {
		<-ctx.Done()
		if cmd.Process != nil {
			cmd.Process.Kill()
		}
	}()

	jParser := NewJournalEntryParser(sm.config.Name, sm.config.Service, sm.hostname, sm.drainConfig, sm.stateDir)
	defer jParser.Close()

	for scanner.Scan() {
		select {
		case <-ctx.Done():
			return nil
		default:
		}

		line := scanner.Text()
		if strings.TrimSpace(line) == "" {
			continue
		}

		entry, err := jParser.Parse(line)
		if err != nil {
			slog.Error("error parsing journal entry", "service", sm.config.Name, "error", err)
			continue
		}

		select {
		case out <- entry:
		case <-ctx.Done():
			return nil
		default:
			slog.Warn("Service-Log-Channel is full, entry dropped", "service", sm.config.Name)
		}
	}

	if err := scanner.Err(); err != nil {
		return fmt.Errorf("scanner error: %w", err)
	}

	return cmd.Wait()
}

func (sm *ServiceMonitor) buildJournalctlArgs() []string {
	args := []string{
		"sudo",
		"journalctl",
		"-f",
		"-o", "json",
		"-u", sm.config.Service,
	}

	if sm.config.SinceTime != "" {
		args = append(args, "--since", sm.config.SinceTime)
	}

	if sm.config.Priority != "" {
		args = append(args, "-p", sm.config.Priority)
	}

	return args
}

type JournalEntryParser struct {
	serviceName string
	unitName    string
	hostname    string
	innerParser parser.Parser
}

func NewJournalEntryParser(serviceName, unitName, hostname string, drainCfg *drain3.Config, stateDir string) *JournalEntryParser {
	pCfg := parser.ParserConfig{
		ServiceName: serviceName,
		LogType:     "custom",
		Hostname:    hostname,
		DrainConfig: drainCfg,
		StateDir:    stateDir,
	}

	inner, err := parser.New(pCfg)
	if err != nil {
		slog.Error("Failed to create inner parser for service", "service", serviceName, "error", err)
	}

	return &JournalEntryParser{
		serviceName: serviceName,
		unitName:    unitName,
		hostname:    hostname,
		innerParser: inner,
	}
}

func (jep *JournalEntryParser) Close() error {
	if jep.innerParser != nil {
		return jep.innerParser.Close()
	}
	return nil
}

func (jep *JournalEntryParser) Parse(jsonLine string) (models.LogMessage, error) {
	var journalData map[string]any
	if err := json.Unmarshal([]byte(jsonLine), &journalData); err != nil {
		return models.LogMessage{}, fmt.Errorf("JSON unmarshal error: %w", err)
	}

	entry := models.NewLogMessage("service_log", jep.hostname)
	entry.Service = jep.serviceName
	entry.Unit = jep.unitName

	if tsStr, ok := journalData["__REALTIME_TIMESTAMP"].(string); ok {
		if tsInt, err := strconv.ParseInt(tsStr, 10, 64); err == nil {
			entry.Timestamp = time.Unix(0, tsInt*1000)
		}
	}
	if entry.Timestamp.IsZero() {
		entry.Timestamp = time.Now()
	}

	if msg, ok := journalData["MESSAGE"].(string); ok {
		entry.LogMessage = msg
	}

	if priority, ok := journalData["PRIORITY"].(string); ok {
		entry.Priority = priority
		entry.PriorityName = jep.getPriorityName(priority)
	}

	if pidStr, ok := journalData["_PID"].(string); ok {
		if pid, err := strconv.Atoi(pidStr); err == nil {
			entry.PID = pid
		}
	}

	jep.extractSystemdFields(journalData, &entry)

	if bootID, ok := journalData["_BOOT_ID"].(string); ok {
		entry.BootID = bootID
	}
	if machineID, ok := journalData["_MACHINE_ID"].(string); ok {
		entry.MachineID = machineID
	}
	if hostname, ok := journalData["_HOSTNAME"].(string); ok {
		entry.Host = hostname
	}

	entry.Raw = jsonLine

	if jep.innerParser != nil && entry.LogMessage != "" {
		parsedMsg, err := jep.innerParser.Parse(entry.LogMessage)
		if err == nil {
			jep.mergeEntries(&entry, &parsedMsg)
		}
	}

	return entry, nil
}

func (jep *JournalEntryParser) mergeEntries(target *models.LogMessage, source *models.LogMessage) {
	for k, v := range source.Fields {
		target.Fields[k] = v
	}
	if source.LogLevel != "" {
		target.LogLevel = source.LogLevel
	}
}

func (jep *JournalEntryParser) getPriorityName(priority string) string {
	priorityNames := map[string]string{
		"0": "emergency",
		"1": "alert",
		"2": "critical",
		"3": "error",
		"4": "warning",
		"5": "notice",
		"6": "info",
		"7": "debug",
	}

	if name, exists := priorityNames[priority]; exists {
		return name
	}
	return "unknown"
}

func (jep *JournalEntryParser) extractSystemdFields(journalData map[string]any, entry *models.LogMessage) {
	systemdFields := []string{
		"_SYSTEMD_UNIT", "_SYSTEMD_USER_UNIT", "_SYSTEMD_SLICE",
		"_BOOT_ID", "_MACHINE_ID", "_HOSTNAME", "_TRANSPORT",
		"_CAP_EFFECTIVE", "_SELINUX_CONTEXT", "_AUDIT_SESSION",
		"_AUDIT_LOGINUID", "_GID", "_UID", "_COMM", "_EXE",
		"_CMDLINE", "_SYSTEMD_CGROUP", "_SYSTEMD_SESSION",
		"_SYSTEMD_OWNER_UID", "_SOURCE_REALTIME_TIMESTAMP",
	}

	for _, field := range systemdFields {
		if value, ok := journalData[field]; ok {
			esFieldName := strings.ToLower(strings.TrimPrefix(field, "_"))
			if entry.Fields == nil {
				entry.Fields = make(map[string]any)
			}
			entry.Fields[esFieldName] = value
		}
	}
}