refactor: use parser package for file monitor
This commit is contained in:
parent
e468b3a0e3
commit
72b6ad88c7
2 changed files with 81 additions and 133 deletions
150
file_monitor.go
150
file_monitor.go
|
|
@ -5,49 +5,44 @@ import (
|
|||
"fmt"
|
||||
"log/slog"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"tixel_watch/helpers"
|
||||
"tixel_watch/models"
|
||||
"tixel_watch/parser"
|
||||
|
||||
"github.com/hpcloud/tail"
|
||||
)
|
||||
|
||||
type FileMonitor struct {
|
||||
config ToolConfig
|
||||
parser LogParser
|
||||
}
|
||||
|
||||
type LogParser interface {
|
||||
Parse(line string, toolName string) models.LogMessage
|
||||
parser parser.Parser
|
||||
}
|
||||
|
||||
func NewFileMonitor(config ToolConfig) *FileMonitor {
|
||||
var parser LogParser
|
||||
var logParser parser.Parser
|
||||
|
||||
if config.Format.Pattern != "" {
|
||||
pattern, err := regexp.Compile(config.Format.Pattern)
|
||||
if err != nil {
|
||||
slog.Error("invalid regex pattern", "tool", config.Name, "error", err)
|
||||
parser = &DefaultLogParser{}
|
||||
logParser = &parser.DefaultParser{}
|
||||
} else {
|
||||
parser = &RegexLogParser{
|
||||
pattern: pattern,
|
||||
fields: config.Format.Fields,
|
||||
logParser = &parser.RegexLogParser{
|
||||
Pattern: pattern,
|
||||
Fields: config.Format.Fields,
|
||||
Toolname: config.Name,
|
||||
}
|
||||
}
|
||||
} else {
|
||||
switch config.Name {
|
||||
case "nginx-tjm":
|
||||
parser = &NginxTJMLogParser{}
|
||||
default:
|
||||
parser = &DefaultLogParser{}
|
||||
var err error
|
||||
logParser, err = parser.New(config.Name, "custom")
|
||||
if err != nil {
|
||||
slog.Error("cannot get tool specific parser", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
return &FileMonitor{
|
||||
config: config,
|
||||
parser: parser,
|
||||
parser: logParser,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -85,7 +80,10 @@ func (fm *FileMonitor) Start(ctx context.Context, out chan<- models.LogMessage)
|
|||
continue
|
||||
}
|
||||
|
||||
entry := fm.parser.Parse(line.Text, fm.config.Name)
|
||||
entry, err := fm.parser.Parse(line.Text)
|
||||
if err != nil {
|
||||
slog.Error("error parsing log line", "error", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case out <- entry:
|
||||
|
|
@ -97,117 +95,3 @@ func (fm *FileMonitor) Start(ctx context.Context, out chan<- models.LogMessage)
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
type DefaultLogParser struct{}
|
||||
|
||||
func (p *DefaultLogParser) Parse(line string, toolName string) models.LogMessage {
|
||||
entry := models.NewLogMessage("log_entry", hostname)
|
||||
entry.Tool = toolName
|
||||
entry.LogMessage = strings.TrimSpace(line)
|
||||
entry.Raw = line
|
||||
return entry
|
||||
}
|
||||
|
||||
type RegexLogParser struct {
|
||||
pattern *regexp.Regexp
|
||||
fields map[string]string
|
||||
}
|
||||
|
||||
func (p *RegexLogParser) Parse(line string, toolName string) models.LogMessage {
|
||||
entry := models.NewLogMessage("log_entry", hostname)
|
||||
entry.Tool = toolName
|
||||
entry.Raw = line
|
||||
|
||||
fields := p.parseWithPattern(line)
|
||||
if fields != nil {
|
||||
entry.Fields = fields
|
||||
} else {
|
||||
entry.LogMessage = strings.TrimSpace(line)
|
||||
}
|
||||
|
||||
return entry
|
||||
}
|
||||
|
||||
func (p *RegexLogParser) parseWithPattern(text string) map[string]any {
|
||||
matches := p.pattern.FindStringSubmatch(text)
|
||||
if matches == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
fields := make(map[string]any)
|
||||
subexpNames := p.pattern.SubexpNames()
|
||||
|
||||
for i, match := range matches {
|
||||
if i == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
if i < len(subexpNames) && subexpNames[i] != "" {
|
||||
fieldName := subexpNames[i]
|
||||
|
||||
if mappedName, exists := p.fields[fieldName]; exists {
|
||||
fieldName = mappedName
|
||||
}
|
||||
|
||||
fields[fieldName] = match
|
||||
}
|
||||
}
|
||||
|
||||
return fields
|
||||
}
|
||||
|
||||
type NginxTJMLogParser struct{}
|
||||
|
||||
func (p *NginxTJMLogParser) Parse(line string, toolName string) models.LogMessage {
|
||||
entry := models.NewLogMessage("log_entry", hostname)
|
||||
entry.Tool = toolName
|
||||
entry.Raw = line
|
||||
entry = p.parseNginxTJM(entry)
|
||||
return entry
|
||||
}
|
||||
|
||||
func (p *NginxTJMLogParser) parseNginxTJM(entry models.LogMessage) models.LogMessage {
|
||||
newEntry := entry
|
||||
var nginxBase models.NGinXBaseInfo
|
||||
parts := strings.Fields(entry.Raw)
|
||||
if len(parts) < 10 {
|
||||
return newEntry
|
||||
}
|
||||
|
||||
if len(parts) > 0 {
|
||||
timestampStr := strings.Trim(parts[0], "[]")
|
||||
timestamp, err := helpers.ParseRFC3339WithOptionalZ(timestampStr)
|
||||
if err != nil {
|
||||
slog.Error("unable to parse time", "error", err)
|
||||
}
|
||||
newEntry.Timestamp = timestamp
|
||||
}
|
||||
|
||||
if len(parts) > 2 {
|
||||
nginxBase.ClientIP = parts[2]
|
||||
}
|
||||
|
||||
for i, part := range parts {
|
||||
if strings.HasPrefix(part, "\"") {
|
||||
if i+1 < len(parts) {
|
||||
nginxBase.HTTPMethod = strings.Trim(part, "\"")
|
||||
nginxBase.Route = parts[i+1]
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
for _, part := range parts {
|
||||
if after, ok := strings.CutPrefix(part, "status="); ok {
|
||||
statusCode, err := strconv.ParseInt(after, 10, 64)
|
||||
if err != nil {
|
||||
slog.Error("cant convert statuscode", "error", err)
|
||||
}
|
||||
nginxBase.StatusCode = int(statusCode)
|
||||
break
|
||||
}
|
||||
}
|
||||
newEntry.ServiceInformation = nginxBase
|
||||
|
||||
return newEntry
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue