PMM-9875 Errors view usage.

agent/agents/postgres/pgstatmonitor/models.go

@@ -50,6 +50,16 @@ type pgStatMonitorSettingsTextValue struct {
 	Value string `reform:"value"`
 }
 
+// pgStatMonitorErrors represents a row in pg_stat_monitor_errors view.
+//
+//reform:pg_stat_monitor_errors
+type pgStatMonitorErrors struct {
+	Severity    string `reform:"severity"`
+	Message     string `reform:"message"`
+	MessageTime string `reform:"msgtime"`
+	Calls       int64  `reform:"calls"`
+}
+
 // pgStatMonitorExtended contains pgStatMonitor data and extends it with database, username and tables data.
 // It's made for performance reason.
 type pgStatMonitorExtended struct {

agent/agents/postgres/pgstatmonitor/pgstatmonitor.go

@@ -50,6 +50,7 @@ type PGStatMonitorQAN struct {
 	changes              chan agents.Change
 	monitorCache         *statMonitorCache
 	disableQueryExamples bool
+	waitTime             time.Duration
 }
 
 // Params represent Agent parameters.
@@ -254,6 +255,8 @@ func (m *PGStatMonitorQAN) Run(ctx context.Context) {
 	}
 	running = m.checkDefaultWaitTime(waitTime)
 
+	m.waitTime = waitTime
+
 	// query pg_stat_monitor every waitTime seconds
 	start := time.Now()
 	m.l.Debugf("Scheduling next collection in %s at %s.", waitTime, start.Add(waitTime).Format("15:04:05"))
@@ -299,6 +302,8 @@ func (m *PGStatMonitorQAN) Run(ctx context.Context) {
 				continue
 			}
 
+			m.waitTime = waitTime
+
 			lengthS := uint32(waitTime.Seconds())
 			buckets, err := m.getNewBuckets(ctx, lengthS, normalizedQuery)
 
@@ -407,6 +412,11 @@ func (m *PGStatMonitorQAN) getNewBuckets(ctx context.Context, periodLengthSecs u
 		return nil, err
 	}
 
+	err = m.checkErrorsView(ctx)
+	if err != nil {
+		return nil, err
+	}
+
 	buckets := m.makeBuckets(current, prev)
 	m.l.Debugf("Made %d buckets out of %d stat monitor in %d interval.",
 		len(buckets), len(current), periodLengthSecs)
@@ -426,6 +436,56 @@ func (m *PGStatMonitorQAN) getNewBuckets(ctx context.Context, periodLengthSecs u
 	return buckets, nil
 }
 
+func (m *PGStatMonitorQAN) checkErrorsView(ctx context.Context) error {
+	vPGSM, _, err := getPGMonitorVersion(m.q)
+	if err != nil {
+		return errors.Wrap(err, "failed to get row and view for pg_stat_monitor version")
+	}
+
+	// Errors view is supported in PGSM 2.0 and higher.
+	if vPGSM < pgStatMonitorVersion20PG12 {
+		return nil
+	}
+
+	row := &pgStatMonitorErrors{}
+	rows, err := m.q.SelectRows(pgStatMonitorErrorsView, "")
+	if err != nil {
+		return errors.Wrap(err, "failed to query pg_stat_monitor_errors view")
+	}
+
+	now := time.Now()
+	for ctx.Err() == nil {
+		if err = m.q.NextRow(row, rows); err != nil {
+			if errors.Is(err, reform.ErrNoRows) {
+				break
+			}
+
+			return errors.Wrap(err, "cannot read row from errors view")
+		}
+
+		messageTime, err := time.Parse("2006-01-02 15:04:05", row.MessageTime)
+		if err != nil {
+			return errors.Wrap(err, "cannot parse messageTime")
+		}
+
+		if now.After(messageTime.Add(m.waitTime)) {
+			continue
+		}
+
+		template := "Message: %s, Calls: %d"
+		switch row.Severity {
+		case "INFO":
+			m.l.Infof(template, row.Message, row.Calls)
+		case "WARNING":
+			m.l.Warningf(template, row.Message, row.Calls)
+		case "ERROR":
+			m.l.Errorf(template, row.Message, row.Calls)
+		}
+	}
+
+	return nil
+}
+
 // makeBuckets uses current state of pg_stat_monitor table and accumulated previous state
 // to make metrics buckets.
 func (m *PGStatMonitorQAN) makeBuckets(current, cache map[time.Time]map[string]*pgStatMonitorExtended) []*agentpb.MetricsBucket {