1 package reaper
2
3 import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "path/filepath"
9 "regexp"
10 "strings"
11 "time"
12
13 "github.com/cybertec-postgresql/pgwatch/v6/internal/db"
14 "github.com/cybertec-postgresql/pgwatch/v6/internal/log"
15 "github.com/cybertec-postgresql/pgwatch/v6/internal/metrics"
16 "github.com/cybertec-postgresql/pgwatch/v6/internal/sources"
17 "github.com/jackc/pgx/v5"
18 )
19
20
21 var pgSeverities = [...]string{"DEBUG", "INFO", "NOTICE", "WARNING", "ERROR", "LOG", "FATAL", "PANIC"}
22 var pgSeveritiesLocale = map[string]map[string]string{
23 "C.": {"DEBUG": "DEBUG", "LOG": "LOG", "INFO": "INFO", "NOTICE": "NOTICE", "WARNING": "WARNING", "ERROR": "ERROR", "FATAL": "FATAL", "PANIC": "PANIC"},
24 "de": {"DEBUG": "DEBUG", "LOG": "LOG", "INFO": "INFO", "HINWEIS": "NOTICE", "WARNUNG": "WARNING", "FEHLER": "ERROR", "FATAL": "FATAL", "PANIK": "PANIC"},
25 "fr": {"DEBUG": "DEBUG", "LOG": "LOG", "INFO": "INFO", "NOTICE": "NOTICE", "ATTENTION": "WARNING", "ERREUR": "ERROR", "FATAL": "FATAL", "PANIK": "PANIC"},
26 "it": {"DEBUG": "DEBUG", "LOG": "LOG", "INFO": "INFO", "NOTIFICA": "NOTICE", "ATTENZIONE": "WARNING", "ERRORE": "ERROR", "FATALE": "FATAL", "PANICO": "PANIC"},
27 "ko": {"디버그": "DEBUG", "로그": "LOG", "정보": "INFO", "알림": "NOTICE", "경고": "WARNING", "오류": "ERROR", "치명적오류": "FATAL", "손상": "PANIC"},
28 "pl": {"DEBUG": "DEBUG", "DZIENNIK": "LOG", "INFORMACJA": "INFO", "UWAGA": "NOTICE", "OSTRZEŻENIE": "WARNING", "BŁĄD": "ERROR", "KATASTROFALNY": "FATAL", "PANIKA": "PANIC"},
29 "ru": {"ОТЛАДКА": "DEBUG", "СООБЩЕНИЕ": "LOG", "ИНФОРМАЦИЯ": "INFO", "ЗАМЕЧАНИЕ": "NOTICE", "ПРЕДУПРЕЖДЕНИЕ": "WARNING", "ОШИБКА": "ERROR", "ВАЖНО": "FATAL", "ПАНИКА": "PANIC"},
30 "sv": {"DEBUG": "DEBUG", "LOGG": "LOG", "INFO": "INFO", "NOTIS": "NOTICE", "VARNING": "WARNING", "FEL": "ERROR", "FATALT": "FATAL", "PANIK": "PANIC"},
31 "tr": {"DEBUG": "DEBUG", "LOG": "LOG", "BİLGİ": "INFO", "NOT": "NOTICE", "UYARI": "WARNING", "HATA": "ERROR", "ÖLÜMCÜL (FATAL)": "FATAL", "KRİTİK": "PANIC"},
32 "zh": {"调试": "DEBUG", "日志": "LOG", "信息": "INFO", "注意": "NOTICE", "警告": "WARNING", "错误": "ERROR", "致命错误": "FATAL", "比致命错误还过分的错误": "PANIC"},
33 }
34
35 const csvLogDefaultRegEx = `^^(?P<log_time>.*?),"?(?P<user_name>.*?)"?,"?(?P<database_name>.*?)"?,(?P<process_id>\d+),"?(?P<connection_from>.*?)"?,(?P<session_id>.*?),(?P<session_line_num>\d+),"?(?P<command_tag>.*?)"?,(?P<session_start_time>.*?),(?P<virtual_transaction_id>.*?),(?P<transaction_id>.*?),(?P<error_severity>\w+),`
36 const csvLogDefaultGlobSuffix = "*.csv"
37
38 const maxChunkSize uint64 = 10 * 1024 * 1024
39 const maxTrackedFiles = 2500
40
41 type LogParser struct {
42 *LogConfig
43 ctx context.Context
44 LogsMatchRegex *regexp.Regexp
45 SourceConn *sources.DbConn
46 realDbname string
47 Interval time.Duration
48 StoreCh chan<- metrics.MeasurementEnvelope
49 eventCounts map[string]int64
50 eventCountsTotal map[string]int64
51 lastSendTime time.Time
52 fileOffsets map[string]uint64
53 }
54
55 type LogConfig struct {
56 CollectorEnabled bool
57 CSVDestination bool
58 TruncateOnRotation bool
59 Directory string
60 ServerMessagesLang string
61 }
62
63 func NewLogParser(ctx context.Context, mdb *sources.DbConn, storeCh chan<- metrics.MeasurementEnvelope) (lp *LogParser, err error) {
64
65 logger := log.GetLogger(ctx).WithField("source", mdb.Name).WithField("metric", specialMetricServerLogEventCounts)
66 ctx = log.WithLogger(ctx, logger)
67
68 logsRegex := regexp.MustCompile(csvLogDefaultRegEx)
69
70 logger.Debugf("Using %s as log parsing regex", logsRegex)
71
72 var cfg *LogConfig
73 if cfg, err = tryDetermineLogSettings(ctx, mdb.Conn); err != nil {
74 return nil, fmt.Errorf("could not determine Postgres logs settings: %w", err)
75 }
76
77 if !cfg.CollectorEnabled {
78 return nil, errors.New("logging_collector is not enabled on the db server")
79 }
80
81 if !cfg.CSVDestination {
82 return nil, errors.New("log_destination must contain 'csvlog' for log parsing to work")
83 }
84
85 logger.Debugf("Considering log files in folder: %s", cfg.Directory)
86
87 mdb.RLock()
88 realDbname := mdb.RealDbname
89 mdb.RUnlock()
90 return &LogParser{
91 ctx: ctx,
92 LogsMatchRegex: logsRegex,
93 SourceConn: mdb,
94 realDbname: realDbname,
95 Interval: mdb.GetMetricInterval(specialMetricServerLogEventCounts),
96 StoreCh: storeCh,
97 LogConfig: cfg,
98 eventCounts: make(map[string]int64),
99 eventCountsTotal: make(map[string]int64),
100 fileOffsets: make(map[string]uint64),
101 }, nil
102 }
103
104 func (lp *LogParser) HasSendIntervalElapsed() bool {
105 return lp.lastSendTime.IsZero() || lp.lastSendTime.Before(time.Now().Add(-lp.Interval))
106 }
107
108 func (lp *LogParser) ParseLogs() error {
109 l := log.GetLogger(lp.ctx)
110 if ok, err := db.IsClientOnSameHost(lp.SourceConn.Conn); ok && err == nil {
111 l.Info("DB is on the same host, parsing logs locally")
112 if err = checkHasLocalPrivileges(lp.Directory); err == nil {
113 return lp.parseLogsLocal()
114 }
115 l.WithError(err).Error("Couldn't parse logs locally, lacking required privileges")
116 }
117
118 l.Info("DB is not detected to be on the same host, parsing logs remotely")
119 if err := checkHasRemotePrivileges(lp.ctx, lp.SourceConn, lp.Directory); err != nil {
120 l.WithError(err).Error("couldn't parse logs remotely, lacking required privileges")
121 return err
122 }
123 return lp.parseLogsRemote()
124 }
125
126 func tryDetermineLogSettings(ctx context.Context, conn db.PgxIface) (cfg *LogConfig, err error) {
127 sql := `select
128 current_setting('logging_collector') = 'on' as is_enabled,
129 strpos(current_setting('log_destination'), 'csvlog') > 0 as csvlog_dest,
130 current_setting('log_truncate_on_rotation') = 'on' as log_trunc,
131 case
132 when current_setting('log_directory') ~ '^(\w:)?\/.+' then current_setting('log_directory')
133 else current_setting('data_directory') || '/' || current_setting('log_directory')
134 end as log_dir,
135 current_setting('lc_messages')::varchar(2) as lc_messages`
136 var res pgx.Rows
137 if res, err = conn.Query(ctx, sql); err == nil {
138 if cfg, err = pgx.CollectOneRow(res, pgx.RowToAddrOfStructByPos[LogConfig]); err == nil {
139 if _, ok := pgSeveritiesLocale[cfg.ServerMessagesLang]; !ok {
140 cfg.ServerMessagesLang = "en"
141 }
142 return cfg, nil
143 }
144 }
145 return nil, err
146 }
147
148 func checkHasRemotePrivileges(ctx context.Context, mdb *sources.DbConn, logsDirPath string) error {
149 var logFile string
150 err := mdb.Conn.QueryRow(ctx, "select name from pg_ls_logdir() limit 1").Scan(&logFile)
151 if err != nil && err != pgx.ErrNoRows {
152 return err
153 }
154
155 var dummy string
156 err = mdb.Conn.QueryRow(ctx, "select pg_read_file($1, 0, 0)", filepath.Join(logsDirPath, logFile)).Scan(&dummy)
157 return err
158 }
159
160 func checkHasLocalPrivileges(logsDirPath string) error {
161 _, err := os.ReadDir(logsDirPath)
162 if err != nil {
163 return err
164 }
165 return nil
166 }
167
168 func severityToEnglish(serverLang, errorSeverity string) string {
169 if serverLang == "en" {
170 return errorSeverity
171 }
172 severityMap := pgSeveritiesLocale[serverLang]
173 severityEn, ok := severityMap[errorSeverity]
174 if !ok {
175 return errorSeverity
176 }
177 return severityEn
178 }
179
180 func (lp *LogParser) regexMatchesToMap(matches []string) map[string]string {
181 result := make(map[string]string)
182 if len(matches) == 0 || lp.LogsMatchRegex == nil {
183 return result
184 }
185 for i, name := range lp.LogsMatchRegex.SubexpNames() {
186 if i != 0 && name != "" {
187 result[name] = matches[i]
188 }
189 }
190 return result
191 }
192
193
194 func (lp *LogParser) GetMeasurementEnvelope() metrics.MeasurementEnvelope {
195 allSeverityCounts := metrics.NewMeasurement(time.Now().UnixNano())
196 for _, s := range pgSeverities {
197 parsedCount, ok := lp.eventCounts[s]
198 if ok {
199 allSeverityCounts[strings.ToLower(s)] = parsedCount
200 } else {
201 allSeverityCounts[strings.ToLower(s)] = int64(0)
202 }
203 parsedCount, ok = lp.eventCountsTotal[s]
204 if ok {
205 allSeverityCounts[strings.ToLower(s)+"_total"] = parsedCount
206 } else {
207 allSeverityCounts[strings.ToLower(s)+"_total"] = int64(0)
208 }
209 }
210 return metrics.MeasurementEnvelope{
211 DBName: lp.SourceConn.Name,
212 MetricName: specialMetricServerLogEventCounts,
213 Data: metrics.Measurements{allSeverityCounts},
214 CustomTags: lp.SourceConn.CustomTags,
215 }
216 }
217
218 func zeroEventCounts(eventCounts map[string]int64) {
219 for _, severity := range pgSeverities {
220 eventCounts[severity] = 0
221 }
222 }
223