@@ -6,12 +6,14 @@ import (
66 "fmt"
77 "time"
88
9+ "github.com/jackc/pgx/v5/pgtype"
910 abd "github.com/qiniu/zeroops/internal/alerting/database"
1011)
1112
1213// PgStore is a PostgreSQL-backed Store implementation using the alerting database wrapper.
1314// Note: The current database wrapper does not expose transactions; WithTx acts as a simple wrapper.
1415// For production-grade atomicity, extend the database wrapper to support sql.Tx and wire it here.
16+ // This implementation uses pgx native types to avoid manual parsing of PostgreSQL interval types.
1517type PgStore struct {
1618 DB * abd.Database
1719}
@@ -76,16 +78,22 @@ func (s *PgStore) DeleteRule(ctx context.Context, name string) error {
7678}
7779
7880func (s * PgStore ) UpsertMeta (ctx context.Context , m * AlertRuleMeta ) (bool , error ) {
79- labelsJSON , _ := json .Marshal (m .Labels )
81+ labelsJSON , err := json .Marshal (m .Labels )
82+ if err != nil {
83+ return false , fmt .Errorf ("marshal labels: %w" , err )
84+ }
85+
86+ // Convert time.Duration to pgtype.Interval
87+ interval := durationToPgInterval (m .WatchTime )
88+
8089 const q = `
8190 INSERT INTO alert_rule_metas(alert_name, labels, threshold, watch_time)
8291 VALUES ($1, $2::jsonb, $3, $4)
8392 ON CONFLICT (alert_name, labels) DO UPDATE SET
8493 threshold=EXCLUDED.threshold,
85- watch_time=EXCLUDED.watch_time,
86- updated_at=now()
94+ watch_time=EXCLUDED.watch_time
8795 `
88- _ , err : =s .DB .ExecContext (ctx , q , m .AlertName , string (labelsJSON ), m .Threshold , m . WatchTime )
96+ _ , err = s .DB .ExecContext (ctx , q , m .AlertName , string (labelsJSON ), m .Threshold , interval )
8997 if err != nil {
9098 return false , fmt .Errorf ("upsert meta: %w" , err )
9199 }
@@ -94,7 +102,10 @@ func (s *PgStore) UpsertMeta(ctx context.Context, m *AlertRuleMeta) (bool, error
94102}
95103
96104func (s * PgStore ) GetMetas (ctx context.Context , name string , labels LabelMap ) ([]* AlertRuleMeta , error ) {
97- labelsJSON , _ := json .Marshal (labels )
105+ labelsJSON , err := json .Marshal (labels )
106+ if err != nil {
107+ return nil , fmt .Errorf ("marshal labels for get: %w" , err )
108+ }
98109 const q = `
99110 SELECT alert_name, labels, threshold, watch_time
100111 FROM alert_rule_metas
@@ -110,61 +121,99 @@ func (s *PgStore) GetMetas(ctx context.Context, name string, labels LabelMap) ([
110121 var alertName string
111122 var labelsRaw string
112123 var threshold float64
113- var watch any
124+ var watch pgtype. Interval
114125 if err := rows .Scan (& alertName , & labelsRaw , & threshold , & watch ); err != nil {
115126 return nil , fmt .Errorf ("scan meta: %w" , err )
116127 }
117128 lm := LabelMap {}
118- _ = json .Unmarshal ([]byte (labelsRaw ), & lm )
129+ if err := json .Unmarshal ([]byte (labelsRaw ), & lm ); err != nil {
130+ return nil , fmt .Errorf ("unmarshal labels: %w" , err )
131+ }
119132 meta := & AlertRuleMeta {AlertName : alertName , Labels : lm , Threshold : threshold }
120- // best-effort: watch_time may come back as string or duration; we try string -> duration
121- switch v := watch .(type ) {
122- case string :
123- if d , err := timeParseDurationPG (v ); err == nil {
124- meta .WatchTime = d
125- }
133+
134+ // Convert pgtype.Interval to time.Duration
135+ if duration , err := pgIntervalToDuration (watch ); err == nil {
136+ meta .WatchTime = duration
126137 }
127138 res = append (res , meta )
128139 }
129140 return res , nil
130141}
131142
132143func (s * PgStore ) DeleteMeta (ctx context.Context , name string , labels LabelMap ) error {
133- labelsJSON , _ := json .Marshal (labels )
144+ labelsJSON , err := json .Marshal (labels )
145+ if err != nil {
146+ return fmt .Errorf ("marshal labels: %w" , err )
147+ }
134148 const q = `DELETE FROM alert_rule_metas WHERE alert_name=$1 AND labels=$2::jsonb`
135- _ , err : =s .DB .ExecContext (ctx , q , name , string (labelsJSON ))
149+ _ , err = s .DB .ExecContext (ctx , q , name , string (labelsJSON ))
136150 if err != nil {
137151 return fmt .Errorf ("delete meta: %w" , err )
138152 }
139153 return nil
140154}
141155
142156func (s * PgStore ) InsertChangeLog (ctx context.Context , log * ChangeLog ) error {
143- labelsJSON , _ := json .Marshal (log .Labels )
157+ labelsJSON , err := json .Marshal (log .Labels )
158+ if err != nil {
159+ return fmt .Errorf ("marshal labels for changelog: %w" , err )
160+ }
161+
162+ // Convert time.Duration to pgtype.Interval for old and new watch times
163+ var oldWatch , newWatch * pgtype.Interval
164+ if log .OldWatch != nil {
165+ interval := durationToPgInterval (* log .OldWatch )
166+ oldWatch = & interval
167+ }
168+ if log .NewWatch != nil {
169+ interval := durationToPgInterval (* log .NewWatch )
170+ newWatch = & interval
171+ }
172+
144173 const q = `
145174 INSERT INTO alert_meta_change_logs(id, alert_name, change_type, labels, old_threshold, new_threshold, old_watch, new_watch, change_time)
146175 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
147176 `
148- _ , err : =s .DB .ExecContext (ctx , q , log .ID , log .AlertName , log .ChangeType , string (labelsJSON ), log .OldThreshold , log .NewThreshold , log . OldWatch , log . NewWatch , log .ChangeTime )
177+ _ , err = s .DB .ExecContext (ctx , q , log .ID , log .AlertName , log .ChangeType , string (labelsJSON ), log .OldThreshold , log .NewThreshold , oldWatch , newWatch , log .ChangeTime )
149178 if err != nil {
150179 return fmt .Errorf ("insert change log: %w" , err )
151180 }
152181 return nil
153182}
154183
155- // timeParseDurationPG parses a small subset of PostgreSQL interval text output into time.Duration.
156- // Supported examples: "01:02:03", "02:03", "3600 seconds". Best-effort only.
157- func timeParseDurationPG (s string ) (time.Duration , error ) {
158- // HH:MM:SS
159- var h , m int
160- var sec float64
161- if n , _ := fmt .Sscanf (s , "%d:%d:%f" , & h , & m , & sec ); n >= 2 {
162- d := time .Duration (h )* time .Hour + time .Duration (m )* time .Minute + time .Duration (sec * float64 (time .Second ))
163- return d , nil
164- }
165- var seconds float64
166- if n , _ := fmt .Sscanf (s , "%f seconds" , & seconds ); n == 1 {
167- return time .Duration (seconds * float64 (time .Second )), nil
168- }
169- return 0 , fmt .Errorf ("unsupported interval format: %s" , s )
184+ // durationToPgInterval converts a time.Duration to pgtype.Interval.
185+ // Note: This conversion assumes the duration represents a fixed time period.
186+ // For durations that include months or years, this conversion may not be accurate.
187+ func durationToPgInterval (d time.Duration ) pgtype.Interval {
188+ // Convert to total microseconds first
189+ totalMicroseconds := d .Microseconds ()
190+
191+ // Calculate days and remaining microseconds
192+ days := totalMicroseconds / (24 * 60 * 60 * 1000000 ) // 24 hours * 60 minutes * 60 seconds * 1,000,000 microseconds
193+ remainingMicroseconds := totalMicroseconds % (24 * 60 * 60 * 1000000 )
194+
195+ return pgtype.Interval {
196+ Microseconds : remainingMicroseconds ,
197+ Days : int32 (days ),
198+ Months : 0 , // Duration doesn't include months
199+ Valid : true ,
200+ }
201+ }
202+
203+ // pgIntervalToDuration converts a pgtype.Interval to time.Duration.
204+ // This function returns an error if the interval contains months or years,
205+ // as these cannot be accurately converted to a fixed duration.
206+ func pgIntervalToDuration (interval pgtype.Interval ) (time.Duration , error ) {
207+ if ! interval .Valid {
208+ return 0 , fmt .Errorf ("interval is not valid" )
209+ }
210+
211+ // Check if the interval contains months or years
212+ if interval .Months != 0 {
213+ return 0 , fmt .Errorf ("cannot convert interval with months to duration: %d months" , interval .Months )
214+ }
215+
216+ // Convert to duration
217+ totalMicroseconds := interval .Microseconds + int64 (interval .Days )* 24 * 60 * 60 * 1000000
218+ return time .Duration (totalMicroseconds ) * time .Microsecond , nil
170219}
0 commit comments