git/watch-tool
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: move models to their own package to use the same as in the importer

Browse source
This commit is contained in:
Patryk Hegenberg 2025-09-24 22:57:37 +02:00
parent 553a85562b
commit 9aa1b7384d
14 changed files with 170 additions and 152 deletions
Download patch file Download diff file Expand all files Collapse all files

11
elasticsearch.go
View file

@ -7,6 +7,7 @@ import (
"log/slog"
"strings"
"time"
"tixel_watch/models"
"github.com/elastic/go-elasticsearch/v8"
)
@ -50,8 +51,8 @@ func TestElasticsearchConnection(es *elasticsearch.Client) error {
}
type ElasticsearchSender interface {
SendBatch(baseIndex string, entries []LogEntry) error
SendSystemMetrics(baseIndex string, metrics SystemResources) error
SendBatch(baseIndex string, entries []models.LogMessage) error
SendSystemMetrics(baseIndex string, metrics models.SystemResources) error
}
type ElasticsearchClient struct {
@ -62,7 +63,7 @@ func NewElasticsearchSender(client *elasticsearch.Client) ElasticsearchSender {
return &ElasticsearchClient{client: client}
}
func (esc *ElasticsearchClient) SendBatch(baseIndex string, entries []LogEntry) error {
func (esc *ElasticsearchClient) SendBatch(baseIndex string, entries []models.LogMessage) error {
if len(entries) == 0 {
return nil
}
@ -105,8 +106,8 @@ func (esc *ElasticsearchClient) SendBatch(baseIndex string, entries []LogEntry)
return nil
}
func (esc *ElasticsearchClient) SendSystemMetrics(baseIndex string, metrics SystemResources) error {
msg := LogEntry{
func (esc *ElasticsearchClient) SendSystemMetrics(baseIndex string, metrics models.SystemResources) error {
msg := models.LogMessage{
Service: "system-metrics",
Timestamp: time.Now(),
LogLevel: "Info",

3
elasticsearch_exporterV2.go
View file

@ -7,6 +7,7 @@ import (
"log/slog"
"strings"
"time"
"tixel_watch/models"
"github.com/elastic/go-elasticsearch/v8"
)
@ -28,7 +29,7 @@ func NewElasticsearchExporterV2(config ElasticsearchConfig) (*ElasticsearchExpor
}, nil
}
func (e *ElasticsearchExporterV2) Export(ctx context.Context, entries []LogEntry) error {
func (e *ElasticsearchExporterV2) Export(ctx context.Context, entries []models.LogMessage) error {
if len(entries) == 0 {
return nil
}

3
export_manager.go
View file

@ -8,6 +8,7 @@ import (
"strings"
"sync"
"time"
"tixel_watch/models"
)
type ExportManager struct {
@ -144,7 +145,7 @@ func (em *ExportManager) exportBatch(ctx context.Context) {
}
}
func (em *ExportManager) exportWithRetry(ctx context.Context, name string, exporter ExporterInterface, entries []LogEntry) error {
func (em *ExportManager) exportWithRetry(ctx context.Context, name string, exporter ExporterInterface, entries []models.LogMessage) error {
var lastErr error
for attempt := 0; attempt <= em.config.RetryAttempts; attempt++ {

21
file_monitor.go
View file

@ -7,6 +7,7 @@ import (
"regexp"
"strconv"
"strings"
"tixel_watch/models"
"github.com/hpcloud/tail"
)
@ -17,7 +18,7 @@ type FileMonitor struct {
}
type LogParser interface {
Parse(line string, toolName string) LogEntry
Parse(line string, toolName string) models.LogMessage
}
func NewFileMonitor(config ToolConfig) *FileMonitor {
@ -49,7 +50,7 @@ func NewFileMonitor(config ToolConfig) *FileMonitor {
}
}
func (fm *FileMonitor) Start(ctx context.Context, out chan<- LogEntry) error {
func (fm *FileMonitor) Start(ctx context.Context, out chan<- models.LogMessage) error {
t, err := tail.TailFile(fm.config.LogFile, tail.Config{
Follow: true,
ReOpen: true,
@ -98,8 +99,8 @@ func (fm *FileMonitor) Start(ctx context.Context, out chan<- LogEntry) error {
type DefaultLogParser struct{}
func (p *DefaultLogParser) Parse(line string, toolName string) LogEntry {
entry := NewLogEntry("log_entry")
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
@ -111,8 +112,8 @@ type RegexLogParser struct {
fields map[string]string
}
func (p *RegexLogParser) Parse(line string, toolName string) LogEntry {
entry := NewLogEntry("log_entry")
func (p *RegexLogParser) Parse(line string, toolName string) models.LogMessage {
entry := models.NewLogMessage("log_entry", hostname)
entry.Tool = toolName
entry.Raw = line
@ -156,17 +157,17 @@ func (p *RegexLogParser) parseWithPattern(text string) map[string]any {
type NginxTJMLogParser struct{}
func (p *NginxTJMLogParser) Parse(line string, toolName string) LogEntry {
entry := NewLogEntry("log_entry")
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 LogEntry) LogEntry {
func (p *NginxTJMLogParser) parseNginxTJM(entry models.LogMessage) models.LogMessage {
newEntry := entry
var nginxBase NGinXBaseInfo
var nginxBase models.NGinXBaseInfo
parts := strings.Fields(entry.Raw)
if len(parts) < 10 {
return newEntry

7
local_storage.go
View file

@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"encoding/json"
"tixel_watch/models"
_ "modernc.org/sqlite"
)
@ -53,7 +54,7 @@ func (s *StorageService) Close() error {
return s.db.Close()
}
func (s *StorageService) SaveLogEntry(ctx context.Context, entry *LogEntry) error {
func (s *StorageService) SaveLogEntry(ctx context.Context, entry *models.LogMessage) error {
fieldsJSON := ""
if entry.Fields != nil {
b, err := json.Marshal(entry.Fields)
@ -118,10 +119,10 @@ func (s *StorageService) SaveLogEntry(ctx context.Context, entry *LogEntry) erro
return err
}
func (s *StorageService) LoadLogEntry(ctx context.Context, id int64) (*LogEntry, error) {
func (s *StorageService) LoadLogEntry(ctx context.Context, id int64) (*models.LogMessage, error) {
row := s.db.QueryRowContext(ctx, "SELECT service, timestamp, type, host, tool, log_level, log_message, raw, priority, priority_name, unit, pid, boot_id, machine_id, fields, service_information, system_metrics, tool_information FROM log_entries WHERE id = ?", id)
var entry LogEntry
var entry models.LogMessage
var fieldsJSON, serviceInfoJSON, systemMetricsJSON, toolInfoJSON string
err := row.Scan(

19
local_storageV2.go
View file

@ -6,6 +6,7 @@ import (
"encoding/json"
"fmt"
"strings"
"tixel_watch/models"
_ "modernc.org/sqlite"
)
@ -80,11 +81,11 @@ func createTables(db *sql.DB) error {
return nil
}
func (s *SQLiteStorage) Store(ctx context.Context, entry *LogEntry) error {
return s.StoreBatch(ctx, []LogEntry{*entry})
func (s *SQLiteStorage) Store(ctx context.Context, entry *models.LogMessage) error {
return s.StoreBatch(ctx, []models.LogMessage{*entry})
}
func (s *SQLiteStorage) StoreBatch(ctx context.Context, entries []LogEntry) error {
func (s *SQLiteStorage) StoreBatch(ctx context.Context, entries []models.LogMessage) error {
if len(entries) == 0 {
return nil
}
@ -139,7 +140,7 @@ func (s *SQLiteStorage) StoreBatch(ctx context.Context, entries []LogEntry) erro
return tx.Commit()
}
func (s *SQLiteStorage) Query(ctx context.Context, query StorageQuery) ([]LogEntry, error) {
func (s *SQLiteStorage) Query(ctx context.Context, query StorageQuery) ([]models.LogMessage, error) {
sqlQuery := "SELECT service, timestamp, type, host, tool, log_level, log_message, raw, priority, priority_name, unit, pid, boot_id, machine_id, fields, service_information, system_metrics, tool_information FROM log_entries WHERE 1=1"
args := []any{}
argCount := 0
@ -198,9 +199,9 @@ func (s *SQLiteStorage) Query(ctx context.Context, query StorageQuery) ([]LogEnt
}
defer rows.Close()
var entries []LogEntry
var entries []models.LogMessage
for rows.Next() {
var entry LogEntry
var entry models.LogMessage
var fieldsJSON, serviceInfoJSON, systemMetricsJSON, toolInfoJSON string
err := rows.Scan(
@ -280,7 +281,7 @@ func (s *SQLiteStorage) MarkAsExported(ctx context.Context, ids []int64) error {
return err
}
func (s *SQLiteStorage) GetUnexportedEntries(ctx context.Context, limit int) ([]LogEntry, error) {
func (s *SQLiteStorage) GetUnexportedEntries(ctx context.Context, limit int) ([]models.LogMessage, error) {
sqlQuery := `SELECT id, service, timestamp, type, host, tool, log_level, log_message, raw, priority, priority_name,
unit, pid, boot_id, machine_id, fields, service_information, system_metrics, tool_information
FROM log_entries WHERE exported_at IS NULL ORDER BY timestamp ASC`
@ -295,9 +296,9 @@ func (s *SQLiteStorage) GetUnexportedEntries(ctx context.Context, limit int) ([]
}
defer rows.Close()
var entries []LogEntry
var entries []models.LogMessage
for rows.Next() {
var entry LogEntry
var entry models.LogMessage
var id int64
var fieldsJSON, serviceInfoJSON, systemMetricsJSON, toolInfoJSON string

7
log_processor.go
View file

@ -4,6 +4,7 @@ import (
"context"
"log/slog"
"time"
"tixel_watch/models"
)
type LogProcessor struct {
@ -18,8 +19,8 @@ func NewLogProcessor(storage StorageInterface) *LogProcessor {
}
}
func (lp *LogProcessor) Start(ctx context.Context, logChan <-chan LogEntry) {
batch := make([]LogEntry, 0, lp.batchSize)
func (lp *LogProcessor) Start(ctx context.Context, logChan <-chan models.LogMessage) {
batch := make([]models.LogMessage, 0, lp.batchSize)
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
@ -58,7 +59,7 @@ func (lp *LogProcessor) Start(ctx context.Context, logChan <-chan LogEntry) {
}
}
func (lp *LogProcessor) storeBatch(ctx context.Context, batch []LogEntry) {
func (lp *LogProcessor) storeBatch(ctx context.Context, batch []models.LogMessage) {
if len(batch) == 0 {
return
}

3
main.go
View file

@ -8,6 +8,7 @@ import (
"sync"
"syscall"
"time"
"tixel_watch/models"
)
var hostname string
@ -76,7 +77,7 @@ func main() {
// exportManager.RegisterExporter("grafana", grafanaExporter)
}
logChan := make(chan LogEntry, 1000)
logChan := make(chan models.LogMessage, 1000)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

117
models.go → models/models.go
View file

@ -1,8 +1,13 @@
package main
package models
import (
"time"
)
import "time"
type ESDocument struct {
Index string `json:"_index"`
Type string `json:"_type,omitempty"`
ID string `json:"_id,omitempty"`
Log LogMessage `json:"log"`
}
type SystemResources struct {
Timestamp time.Time `json:"@timestamp"`
@ -112,7 +117,7 @@ type NetworkStat struct {
PacketsRecv uint64 `json:"packets_recv"`
}
type LogEntry struct {
type LogMessage struct {
Service string `json:"service,omitempty"`
Timestamp time.Time `json:"timestamp"`
Type string `json:"type"`
@ -120,7 +125,6 @@ type LogEntry struct {
Tool string `json:"tool,omitempty"`
LogLevel string `json:"log_level"`
LogMessage string `json:"log_message,omitempty"`
SyslogInfo SyslogFields `json:"syslog_information,omitempty"`
ServiceInformation any `json:"service_info,omitempty"`
SystemMetrics any `json:"system-metrics,omitempty"`
ToolInformation any `json:"tool_info,omitempty"`
@ -132,40 +136,22 @@ type LogEntry struct {
BootID string `json:"boot_id,omitempty"`
MachineID string `json:"machine_id,omitempty"`
Fields map[string]any `json:"fields,omitempty"`
// SyslogInfo SyslogFields `json:"syslog_information,omitempty"`
}
// type LogMessage struct {
// Service string `json:"service"`
// Timestamp time.Time `json:"timestamp"`
// LogLevel string `json:"log_level"`
// LogMessage string `json:"log_message"`
// SyslogInfo SyslogFields `json:"syslog_information"`
// ServiceInformation any `json:"service_info,omitempty"`
// }
type SyslogFields struct {
SysLogTimestamp time.Time `json:"syslog_timestamp,omitempty"`
Hostname string `json:"hostname,omitempty"`
SysLogTimestamp time.Time `json:"syslog_timestamp"`
Hostname string `json:"hostname"`
ProcessInfo string `json:"process_info"`
Raw string `json:"raw,omitempty"`
Priority string `json:"priority,omitempty"`
Unit string `json:"unit,omitempty"`
PID int `json:"pid,omitempty"`
BootID string `json:"boot_id,omitempty"`
MachineID string `json:"machine_id,omitempty"`
Fields map[string]any `json:"fields"`
}
func NewLogEntry(entryType string) LogEntry {
return LogEntry{
Timestamp: time.Now(),
Type: entryType,
Host: hostname,
}
}
func NewSystemResources() SystemResources {
return SystemResources{
Timestamp: time.Now(),
Type: "system_metrics",
Host: hostname,
DiskUsage: make(map[string]DiskUsage),
DiskIOStats: make(map[string]DiskIOStat),
NetworkStats: make(map[string]NetworkStat),
NetworkLatency: make(map[string]LatencyInfo),
BandwidthUtilization: make(map[string]BandwidthInfo),
}
}
type AMBaseInfo struct {
@ -180,25 +166,11 @@ type TCCBaseInfo struct {
LoggerName string `json:"logger_name"`
}
type NGinXBaseInfo struct {
ClientIP string `json:"client_ip"`
RemoteUser string `json:"remote_user"`
Request string `json:"request"`
StatusCode int `json:"status_code"`
BytesSend int `json:"bytes_sent"`
Referer string `json:"referer"`
UserAgent string `json:"user_agent"`
HTTPMethod string `json:"http_method"`
RequestURI string `json:"request_uri"`
HTTPVersion string `json:"http_version"`
Route string `json:"route"`
}
type TSTransferInfo struct {
TransferID string `json:"transfer_identifier,omitempty"`
Lane int `json:"lane,omitempty"`
StartTime time.Time `json:"start_time,omitempty"`
EndTime time.Time `json:"end_time,omitempty"`
StartTime time.Time `json:"start_time"`
EndTime time.Time `json:"end_time"`
Buffers int `json:"buffers,omitempty"`
Streams int `json:"streams,omitempty"`
ChunkSize int `json:"chunksize,omitempty"`
@ -219,6 +191,7 @@ type TSTransferInfo struct {
TheoreticalRate float64 `json:"theoretical_rate_mbs,omitempty"`
Efficiency float64 `json:"efficiency_percent,omitempty"`
Status string `json:"status,omitempty"`
Hostname string `json:"hostname"`
}
type TJMTransferInfo struct {
@ -229,8 +202,8 @@ type TJMTransferInfo struct {
CorrelationID string `json:"correlation_id,omitempty"`
ThreadID string `json:"thread_id,omitempty"`
JavaClass string `json:"java_class,omitempty"`
StartTime time.Time `json:"start_time,omitempty"`
EndTime time.Time `json:"end_time,omitempty"`
StartTime time.Time `json:"start_time"`
EndTime time.Time `json:"end_time"`
TransferLane string `json:"transfer_lane,omitempty"`
Dest string `json:"destination,omitempty"`
Src string `json:"source,omitempty"`
@ -242,4 +215,40 @@ type TJMTransferInfo struct {
DataRateMBs float64 `json:"datarate_mbs,omitempty"`
BytesProcessed int64 `json:"bytes_processed,omitempty"`
FileSizeMB float64 `json:"file_size_mb,omitempty"`
Hostname string `json:"hostname"`
}
type NGinXBaseInfo struct {
ClientIP string `json:"client_ip"`
RemoteUser string `json:"remote_user"`
Request string `json:"request"`
StatusCode int `json:"status_code"`
BytesSend int `json:"bytes_sent"`
Referer string `json:"referer"`
UserAgent string `json:"user_agent"`
HTTPMethod string `json:"http_method"`
RequestURI string `json:"request_uri"`
HTTPVersion string `json:"http_version"`
Route string `json:"route"`
}
func NewLogMessage(entryType, hostname string) LogMessage {
return LogMessage{
Timestamp: time.Now(),
Type: entryType,
Host: hostname,
}
}
func NewSystemResources(hostname string) SystemResources {
return SystemResources{
Timestamp: time.Now(),
Type: "system_metrics",
Host: hostname,
DiskUsage: make(map[string]DiskUsage),
DiskIOStats: make(map[string]DiskIOStat),
NetworkStats: make(map[string]NetworkStat),
NetworkLatency: make(map[string]LatencyInfo),
BandwidthUtilization: make(map[string]BandwidthInfo),
}
}

43
service_monitor.go
View file

@ -11,6 +11,7 @@ import (
"strconv"
"strings"
"time"
"tixel_watch/models"
)
type ServiceMonitor struct {
@ -23,7 +24,7 @@ func NewServiceMonitor(config ServiceConfig) *ServiceMonitor {
}
}
func (sm *ServiceMonitor) Start(ctx context.Context, out chan<- LogEntry) error {
func (sm *ServiceMonitor) Start(ctx context.Context, out chan<- models.LogMessage) error {
args := sm.buildJournalctlArgs()
slog.Info("starting journalctl", "arguments", args)
@ -116,13 +117,13 @@ func NewJournalEntryParser(serviceName, unitName string) *JournalEntryParser {
}
}
func (jep *JournalEntryParser) Parse(jsonLine string) (LogEntry, error) {
func (jep *JournalEntryParser) Parse(jsonLine string) (models.LogMessage, error) {
var journalData map[string]any
if err := json.Unmarshal([]byte(jsonLine), &journalData); err != nil {
return LogEntry{}, fmt.Errorf("JSON unmarshal error: %w", err)
return models.LogMessage{}, fmt.Errorf("JSON unmarshal error: %w", err)
}
entry := NewLogEntry("service_log")
entry := models.NewLogMessage("service_log", hostname)
entry.Service = jep.serviceName
entry.Unit = jep.unitName
@ -147,7 +148,6 @@ func (jep *JournalEntryParser) Parse(jsonLine string) (LogEntry, error) {
if pidStr, ok := journalData["_PID"].(string); ok {
if pid, err := strconv.Atoi(pidStr); err == nil {
entry.PID = pid
entry.SyslogInfo.ProcessInfo = strconv.FormatInt(int64(pid), 10)
}
}
@ -188,7 +188,7 @@ func (jep *JournalEntryParser) getPriorityName(priority string) string {
return "unknown"
}
func (jep *JournalEntryParser) extractSystemdFields(journalData map[string]any, entry *LogEntry) {
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",
@ -201,15 +201,12 @@ func (jep *JournalEntryParser) extractSystemdFields(journalData map[string]any,
for _, field := range systemdFields {
if value, ok := journalData[field]; ok {
esFieldName := strings.ToLower(strings.TrimPrefix(field, "_"))
if entry.SyslogInfo.Fields == nil {
entry.SyslogInfo.Fields = make(map[string]any)
}
entry.SyslogInfo.Fields[esFieldName] = value
entry.Fields[esFieldName] = value
}
}
}
func (jep *JournalEntryParser) parseServiceSpecific(entry LogEntry) LogEntry {
func (jep *JournalEntryParser) parseServiceSpecific(entry models.LogMessage) models.LogMessage {
switch jep.serviceName {
case "tixstream":
return parseTixstreamService(entry)
@ -230,7 +227,7 @@ var (
amServicePattern = regexp.MustCompile(`^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)\s+(\w+)\s+(\d+)\s+---\s+\[\s*([^\]]*)\]\s+([\w\.]+)\s*:\s*(.*)$`)
tccServicePattern = regexp.MustCompile(`^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)\s+(\w+)\s+(\d+)\s+---\s+\[\s*([^\]]*)\]\s+([\w\.]+)\s*:\s*(.*)$`)
tjmServicePattern = regexp.MustCompile(`^(?<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3})\s+(?<level>\S+)\s+(?<pid>\d+).*?\[(?<collatation_id>[^\]]*)\]\s+\[(?<username>[^\]]*)\]\s+\[(?<thread>[^\]]*)\]\s+(?<class>.*?)\s+:\s+(?<message>.*)`)
tjmTransferNamePattern = regexp.MustCompile(`^(\d{8}T\d{6}-[A-Za-z0-9]+-[A-Za-z]+-(?:in|out)) ?: (.*)$`)
tjmTransferNamePattern = regexp.MustCompile(`^(\d{8}T\d{6}-[A-Za-z0-9]+-.+?-(?:in|out)) ?: (.*)$`)
tsServicePattern = regexp.MustCompile(`^(?<level>\S+)\s+(?<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{6})\s+(?<message>.*)`)
tsTransferIDPattern = regexp.MustCompile(`^(?<transfer>\w{8}-\w{4}-\w{4}-\w{4}-\w{12})\s+(?<message>.*)`)
tjmTransferIDPattern1 = regexp.MustCompile(`(?P<transfer>\w{8}-\w{4}-\w{4}-\w{4}-\w{12}).*?(?P<message>.*)`)
@ -242,9 +239,9 @@ var (
nginxAccessPattern = regexp.MustCompile(`^(\S+)\s+\S+\s+(\S+)\s+\[([^\]]+)\]\s+"([^"]+)"\s+(\d+)\s+(\d+|-)\s*(?:"([^"]*)"\s+"([^"]*)")?`)
)
func parseTixstreamService(entry LogEntry) LogEntry {
func parseTixstreamService(entry models.LogMessage) models.LogMessage {
newEntry := entry
var baseInfo TSTransferInfo
var baseInfo models.TSTransferInfo
matches := tsServicePattern.FindStringSubmatch(newEntry.LogMessage)
if len(matches) > 0 {
@ -349,9 +346,9 @@ func parseTixstreamService(entry LogEntry) LogEntry {
return newEntry
}
func parseTJMService(entry LogEntry) LogEntry {
func parseTJMService(entry models.LogMessage) models.LogMessage {
newEntry := entry
var baseInfo TJMTransferInfo
var baseInfo models.TJMTransferInfo
logContent := entry.LogMessage
msg := strings.TrimSpace(logContent)
@ -375,7 +372,7 @@ func parseTJMService(entry LogEntry) LogEntry {
}
newEntry.LogLevel = strings.TrimSpace(matches[2])
newEntry.LogMessage = strings.TrimSpace(matches[8])
baseInfo = TJMTransferInfo{
baseInfo = models.TJMTransferInfo{
ProcessID: strings.TrimSpace(matches[3]),
CorrelationID: strings.TrimSpace(matches[4]),
Username: strings.TrimSpace(matches[5]),
@ -420,7 +417,7 @@ func parseTJMService(entry LogEntry) LogEntry {
return newEntry
}
func parseAMService(entry LogEntry) LogEntry {
func parseAMService(entry models.LogMessage) models.LogMessage {
newEntry := entry
logContent := newEntry.LogMessage
@ -436,7 +433,7 @@ func parseAMService(entry LogEntry) LogEntry {
}
newEntry.Timestamp = timeParsed
}
baseInfo := AMBaseInfo{
baseInfo := models.AMBaseInfo{
ProcessID: matches[3],
ThreadID: strings.TrimSpace(matches[4]),
LoggerName: matches[5],
@ -448,7 +445,7 @@ func parseAMService(entry LogEntry) LogEntry {
return newEntry
}
func parseTCCService(entry LogEntry) LogEntry {
func parseTCCService(entry models.LogMessage) models.LogMessage {
newEntry := entry
logContent := newEntry.LogMessage
@ -464,7 +461,7 @@ func parseTCCService(entry LogEntry) LogEntry {
}
newEntry.Timestamp = timeParsed
}
baseInfo := TCCBaseInfo{
baseInfo := models.TCCBaseInfo{
ProcessID: matches[3],
ThreadID: strings.TrimSpace(matches[4]),
LoggerName: matches[5],
@ -476,7 +473,7 @@ func parseTCCService(entry LogEntry) LogEntry {
return newEntry
}
func parseNginxService(entry LogEntry) LogEntry {
func parseNginxService(entry models.LogMessage) models.LogMessage {
newEntry := entry
matches := nginxAccessPattern.FindStringSubmatch(strings.TrimSpace(entry.LogMessage))
@ -491,7 +488,7 @@ func parseNginxService(entry LogEntry) LogEntry {
if err != nil {
slog.Error("cant parse bytessend", "error", err)
}
baseInfo := NGinXBaseInfo{
baseInfo := models.NGinXBaseInfo{
ClientIP: matches[1],
RemoteUser: matches[2],
Request: matches[4],

9
storage_interface.go
View file

@ -3,12 +3,13 @@ package main
import (
"context"
"time"
"tixel_watch/models"
)
type StorageInterface interface {
Store(ctx context.Context, entry *LogEntry) error
StoreBatch(ctx context.Context, entries []LogEntry) error
Query(ctx context.Context, query StorageQuery) ([]LogEntry, error)
Store(ctx context.Context, entry *models.LogMessage) error
StoreBatch(ctx context.Context, entries []models.LogMessage) error
Query(ctx context.Context, query StorageQuery) ([]models.LogMessage, error)
Close() error
}
@ -26,6 +27,6 @@ type StorageQuery struct {
}
type ExporterInterface interface {
Export(ctx context.Context, entries []LogEntry) error
Export(ctx context.Context, entries []models.LogMessage) error
HealthCheck(ctx context.Context) error
}

69
system_metrics.go
View file

@ -12,6 +12,7 @@ import (
"strings"
"syscall"
"time"
"tixel_watch/models"
"github.com/elastic/go-elasticsearch/v8"
"github.com/shirou/gopsutil/cpu"
@ -27,8 +28,8 @@ import (
type SystemMetricsCollector struct {
config SystemMetrics
pollInterval int
lastNetworkStats map[string]NetworkStat
lastDiskStats map[string]DiskIOStat
lastNetworkStats map[string]models.NetworkStat
lastDiskStats map[string]models.DiskIOStat
lastMeasureTime time.Time
}
@ -36,8 +37,8 @@ func NewSystemMetricsCollector(config SystemMetrics, pollInterval int) *SystemMe
return &SystemMetricsCollector{
config: config,
pollInterval: pollInterval,
lastNetworkStats: make(map[string]NetworkStat),
lastDiskStats: make(map[string]DiskIOStat),
lastNetworkStats: make(map[string]models.NetworkStat),
lastDiskStats: make(map[string]models.DiskIOStat),
lastMeasureTime: time.Now(),
}
}
@ -67,8 +68,8 @@ func (smc *SystemMetricsCollector) Start(ctx context.Context, es *elasticsearch.
}
}
func (smc *SystemMetricsCollector) collectMetrics() (SystemResources, error) {
result := NewSystemResources()
func (smc *SystemMetricsCollector) collectMetrics() (models.SystemResources, error) {
result := models.NewSystemResources(hostname)
var err error
@ -146,7 +147,7 @@ func (smc *SystemMetricsCollector) collectMetrics() (SystemResources, error) {
return result, nil
}
func (smc *SystemMetricsCollector) collectDiskIOMetrics(result *SystemResources) error {
func (smc *SystemMetricsCollector) collectDiskIOMetrics(result *models.SystemResources) error {
diskIOStats, err := disk.IOCounters()
if err != nil {
return err
@ -155,10 +156,10 @@ func (smc *SystemMetricsCollector) collectDiskIOMetrics(result *SystemResources)
currentTime := time.Now()
timeDiff := currentTime.Sub(smc.lastMeasureTime).Seconds()
result.DiskIOStats = make(map[string]DiskIOStat)
result.DiskIOStats = make(map[string]models.DiskIOStat)
for device, stats := range diskIOStats {
ioStat := DiskIOStat{
ioStat := models.DiskIOStat{
ReadBytes: stats.ReadBytes,
WriteBytes: stats.WriteBytes,
ReadOps: stats.ReadCount,
@ -188,13 +189,13 @@ func (smc *SystemMetricsCollector) collectDiskIOMetrics(result *SystemResources)
return nil
}
func (smc *SystemMetricsCollector) collectNetworkConnections(result *SystemResources) error {
func (smc *SystemMetricsCollector) collectNetworkConnections(result *models.SystemResources) error {
connections, err := psnet.Connections("all")
if err != nil {
return err
}
stats := ConnectionStats{
stats := models.ConnectionStats{
ConnectionsByState: make(map[string]int32),
}
@ -222,7 +223,7 @@ func (smc *SystemMetricsCollector) collectNetworkConnections(result *SystemResou
return nil
}
func (smc *SystemMetricsCollector) collectLoadAverage(result *SystemResources) error {
func (smc *SystemMetricsCollector) collectLoadAverage(result *models.SystemResources) error {
loadAvg, err := load.Avg()
if err != nil {
return err
@ -235,8 +236,8 @@ func (smc *SystemMetricsCollector) collectLoadAverage(result *SystemResources) e
return nil
}
func (smc *SystemMetricsCollector) collectTCPStats(result *SystemResources) error {
tcpStats := TCPStatistics{}
func (smc *SystemMetricsCollector) collectTCPStats(result *models.SystemResources) error {
tcpStats := models.TCPStatistics{}
if data, err := os.ReadFile("/proc/net/netstat"); err == nil {
content := string(data)
@ -251,8 +252,8 @@ func (smc *SystemMetricsCollector) collectTCPStats(result *SystemResources) erro
return nil
}
func (smc *SystemMetricsCollector) collectNetworkLatency(result *SystemResources) error {
result.NetworkLatency = make(map[string]LatencyInfo)
func (smc *SystemMetricsCollector) collectNetworkLatency(result *models.SystemResources) error {
result.NetworkLatency = make(map[string]models.LatencyInfo)
for _, host := range smc.config.LatencyTestHosts {
latency := smc.measureLatency(host)
@ -262,7 +263,7 @@ func (smc *SystemMetricsCollector) collectNetworkLatency(result *SystemResources
return nil
}
func (smc *SystemMetricsCollector) measureLatency(host string) LatencyInfo {
func (smc *SystemMetricsCollector) measureLatency(host string) models.LatencyInfo {
var latencies []time.Duration
var successful int
@ -279,7 +280,7 @@ func (smc *SystemMetricsCollector) measureLatency(host string) LatencyInfo {
}
if len(latencies) == 0 {
return LatencyInfo{Host: host, PacketLoss: 100.0}
return models.LatencyInfo{Host: host, PacketLoss: 100.0}
}
var total time.Duration
@ -300,7 +301,7 @@ func (smc *SystemMetricsCollector) measureLatency(host string) LatencyInfo {
packetLoss := float64(5-successful) / 5.0 * 100.0
jitter := max - min
return LatencyInfo{
return models.LatencyInfo{
Host: host,
MinLatency: min,
MaxLatency: max,
@ -310,13 +311,13 @@ func (smc *SystemMetricsCollector) measureLatency(host string) LatencyInfo {
}
}
func (smc *SystemMetricsCollector) collectBandwidthUsage(result *SystemResources) error {
func (smc *SystemMetricsCollector) collectBandwidthUsage(result *models.SystemResources) error {
netStats, err := psnet.IOCounters(true)
if err != nil {
return err
}
result.BandwidthUtilization = make(map[string]BandwidthInfo)
result.BandwidthUtilization = make(map[string]models.BandwidthInfo)
currentTime := time.Now()
timeDiff := currentTime.Sub(smc.lastMeasureTime).Seconds()
@ -326,7 +327,7 @@ func (smc *SystemMetricsCollector) collectBandwidthUsage(result *SystemResources
continue
}
bandwidth := BandwidthInfo{Interface: stat.Name}
bandwidth := models.BandwidthInfo{Interface: stat.Name}
if lastStat, exists := smc.lastNetworkStats[stat.Name]; exists && timeDiff > 0 {
bytesDiffIn := float64(stat.BytesRecv - lastStat.BytesRecv)
@ -347,7 +348,7 @@ func (smc *SystemMetricsCollector) collectBandwidthUsage(result *SystemResources
}
for _, stat := range netStats {
smc.lastNetworkStats[stat.Name] = NetworkStat{
smc.lastNetworkStats[stat.Name] = models.NetworkStat{
BytesSent: stat.BytesSent,
BytesRecv: stat.BytesRecv,
PacketsSent: stat.PacketsSent,
@ -359,8 +360,8 @@ func (smc *SystemMetricsCollector) collectBandwidthUsage(result *SystemResources
return nil
}
func (smc *SystemMetricsCollector) collectSystemLimits(result *SystemResources) error {
limits := SystemLimitInfo{}
func (smc *SystemMetricsCollector) collectSystemLimits(result *models.SystemResources) error {
limits := models.SystemLimitInfo{}
if data, err := os.ReadFile("/proc/sys/fs/file-max"); err == nil {
if maxFiles, err := strconv.ParseUint(strings.TrimSpace(string(data)), 10, 64); err == nil {
@ -389,13 +390,13 @@ func (smc *SystemMetricsCollector) collectSystemLimits(result *SystemResources)
return nil
}
func (smc *SystemMetricsCollector) collectProcessMetrics(result *SystemResources) error {
func (smc *SystemMetricsCollector) collectProcessMetrics(result *models.SystemResources) error {
processes, err := process.Processes()
if err != nil {
return err
}
var processInfos []ProcessInfo
var processInfos []models.ProcessInfo
var totalOpenFiles int32
for _, p := range processes {
@ -428,7 +429,7 @@ func (smc *SystemMetricsCollector) collectProcessMetrics(result *SystemResources
totalOpenFiles += openFiles
}
processInfos = append(processInfos, ProcessInfo{
processInfos = append(processInfos, models.ProcessInfo{
PID: p.Pid,
Name: name,
CPUPercent: cpuPercent,
@ -453,7 +454,7 @@ func (smc *SystemMetricsCollector) collectProcessMetrics(result *SystemResources
return nil
}
func (smc *SystemMetricsCollector) collectCPUMetrics(result *SystemResources) error {
func (smc *SystemMetricsCollector) collectCPUMetrics(result *models.SystemResources) error {
cpuPercents, err := cpu.Percent(time.Second, false)
if err != nil {
return err
@ -470,7 +471,7 @@ func (smc *SystemMetricsCollector) collectCPUMetrics(result *SystemResources) er
return nil
}
func (smc *SystemMetricsCollector) collectMemoryMetrics(result *SystemResources) error {
func (smc *SystemMetricsCollector) collectMemoryMetrics(result *models.SystemResources) error {
vmStat, err := mem.VirtualMemory()
if err != nil {
return err
@ -483,7 +484,7 @@ func (smc *SystemMetricsCollector) collectMemoryMetrics(result *SystemResources)
return nil
}
func (smc *SystemMetricsCollector) collectDiskMetrics(result *SystemResources) error {
func (smc *SystemMetricsCollector) collectDiskMetrics(result *models.SystemResources) error {
for _, path := range smc.config.DiskPaths {
diskStat, err := disk.Usage(path)
if err != nil {
@ -491,7 +492,7 @@ func (smc *SystemMetricsCollector) collectDiskMetrics(result *SystemResources) e
continue
}
result.DiskUsage[path] = DiskUsage{
result.DiskUsage[path] = models.DiskUsage{
Used: diskStat.Used,
Total: diskStat.Total,
UsedPercent: diskStat.UsedPercent,
@ -502,7 +503,7 @@ func (smc *SystemMetricsCollector) collectDiskMetrics(result *SystemResources) e
return nil
}
func (smc *SystemMetricsCollector) collectNetworkMetrics(result *SystemResources) error {
func (smc *SystemMetricsCollector) collectNetworkMetrics(result *models.SystemResources) error {
netStats, err := psnet.IOCounters(true)
if err != nil {
return err
@ -510,7 +511,7 @@ func (smc *SystemMetricsCollector) collectNetworkMetrics(result *SystemResources
for _, stat := range netStats {
if len(smc.config.NetworkInterfaces) == 0 || slices.Contains(smc.config.NetworkInterfaces, stat.Name) {
result.NetworkStats[stat.Name] = NetworkStat{
result.NetworkStats[stat.Name] = models.NetworkStat{
BytesSent: stat.BytesSent,
BytesRecv: stat.BytesRecv,
PacketsSent: stat.PacketsSent,

5
system_metricsV2.go
View file

@ -4,9 +4,10 @@ import (
"context"
"log/slog"
"time"
"tixel_watch/models"
)
func (smc *SystemMetricsCollector) StartV2(ctx context.Context, storage StorageInterface, logChan chan<- LogEntry) {
func (smc *SystemMetricsCollector) StartV2(ctx context.Context, storage StorageInterface, logChan chan<- models.LogMessage) {
ticker := time.NewTicker(time.Duration(smc.pollInterval) * time.Second)
defer ticker.Stop()
@ -22,7 +23,7 @@ func (smc *SystemMetricsCollector) StartV2(ctx context.Context, storage StorageI
continue
}
entry := NewLogEntry("system_metrics")
entry := models.NewLogMessage("system_metrics", hostname)
entry.Service = "system-metrics"
entry.LogLevel = "Info"
entry.SystemMetrics = metrics

3
web_serviceV2.go
View file

@ -11,6 +11,7 @@ import (
"strconv"
"strings"
"time"
"tixel_watch/models"
)
type WebServiceV2 struct {
@ -212,7 +213,7 @@ func (ws *WebServiceV2) handleStats(w http.ResponseWriter, r *http.Request) {
})
if err != nil {
slog.Error("Failed to query recent entries", "error", err)
recentEntries = []LogEntry{}
recentEntries = []models.LogMessage{}
}
stats := map[string]any{