add PG Materialized View to optimize prometheus aggregator queries

.gitignore

@@ -53,4 +53,5 @@ convoy-ce
 # used to generate test stubs
 .cursor
 
-mise.local.toml
+mise.local.toml
+api/ui/*

cmd/hooks/hooks.go

@@ -716,11 +716,23 @@ func buildCliConfiguration(cmd *cobra.Command) (*config.Configuration, error) {
 					return nil, errors.New("metrics-prometheus-sample-time must be non-zero")
 				}
 
+				queryTimeout, err := cmd.Flags().GetUint64("metrics-prometheus-query-timeout")
+				if err != nil {
+					return nil, err
+				}
+
+				materializedViewRefreshInterval, err := cmd.Flags().GetUint64("metrics-prometheus-materialized-view-refresh-interval")
+				if err != nil {
+					return nil, err
+				}
+
 				c.Metrics = config.MetricsConfiguration{
 					IsEnabled: true,
 					Backend:   config.MetricsBackend(metricsBackend),
 					Prometheus: config.PrometheusMetricsConfiguration{
-						SampleTime: sampleTime,
+						SampleTime:                      sampleTime,
+						QueryTimeout:                    queryTimeout,
+						MaterializedViewRefreshInterval: materializedViewRefreshInterval,
 					},
 				}
 			}

cmd/main.go

@@ -71,6 +71,8 @@ func main() {
 	var dataDogAgentUrl string
 	var metricsBackend string
 	var prometheusMetricsSampleTime uint64
+	var prometheusMetricsQueryTimeout uint64
+	var prometheusMetricsMaterializedViewRefreshInterval uint64
 
 	var retentionPolicy string
 	var retentionPolicyEnabled bool
@@ -143,6 +145,8 @@ func main() {
 	// metrics
 	c.Flags().StringVar(&metricsBackend, "metrics-backend", "prometheus", "Metrics backend e.g. prometheus. ('prometheus' feature flag required")
 	c.Flags().Uint64Var(&prometheusMetricsSampleTime, "metrics-prometheus-sample-time", 5, "Prometheus metrics sample time")
+	c.Flags().Uint64Var(&prometheusMetricsQueryTimeout, "metrics-prometheus-query-timeout", 30, "Prometheus metrics query timeout in seconds")
+	c.Flags().Uint64Var(&prometheusMetricsMaterializedViewRefreshInterval, "metrics-prometheus-materialized-view-refresh-interval", 2, "Materialized view refresh interval in minutes")
 
 	c.Flags().StringVar(&retentionPolicy, "retention-policy", "", "Retention Policy Duration")
 	c.Flags().BoolVar(&retentionPolicyEnabled, "retention-policy-enabled", false, "Retention Policy Enabled")

cmd/server/server.go

@@ -175,6 +175,22 @@ func StartConvoyServer(a *cli.App) error {
 	s.RegisterTask("30 * * * *", convoy.ScheduleQueue, convoy.MonitorTwitterSources)
 	s.RegisterTask("0 * * * *", convoy.ScheduleQueue, convoy.TokenizeSearch)
 
+	// Refresh metrics materialized views
+	if cfg.Metrics.IsEnabled && cfg.Metrics.Backend == config.PrometheusMetricsProvider {
+		refreshInterval := cfg.Metrics.Prometheus.MaterializedViewRefreshInterval
+		if refreshInterval < 1 {
+			refreshInterval = 1
+		}
+		if refreshInterval > 60 {
+			refreshInterval = 60
+		}
+
+		cronSpec := fmt.Sprintf("*/%d * * * *", refreshInterval)
+		s.RegisterTask(cronSpec, convoy.ScheduleQueue, convoy.RefreshMetricsMaterializedViews)
+
+		lo.Infof("Registered metrics materialized view refresh every %d min", refreshInterval)
+	}
+
 	// ensures that project data is backed up about 2 hours before they are deleted
 	if a.Licenser.RetentionPolicy() {
 		// runs at 10pm

cmd/worker/worker.go

@@ -419,6 +419,8 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration) erro
 	consumer.RegisterHandlers(convoy.MetaEventProcessor, task.ProcessMetaEvent(projectRepo, metaEventRepo, dispatcher, a.TracerBackend), nil)
 	consumer.RegisterHandlers(convoy.DeleteArchivedTasksProcessor, task.DeleteArchivedTasks(a.Queue, rd), nil)
 
+	consumer.RegisterHandlers(convoy.RefreshMetricsMaterializedViews, task.RefreshMetricsMaterializedViews(a.DB, rd), nil)
+
 	consumer.RegisterHandlers(convoy.BatchRetryProcessor, task.ProcessBatchRetry(batchRetryRepo, eventDeliveryRepo, a.Queue, lo), nil)
 
 	err = metrics.RegisterQueueMetrics(a.Queue, a.DB, circuitBreakerManager)

config/config.go

@@ -112,7 +112,9 @@ var DefaultConfiguration = Configuration{
 		IsEnabled: false,
 		Backend:   PrometheusMetricsProvider,
 		Prometheus: PrometheusMetricsConfiguration{
-			SampleTime: 5,
+			SampleTime:                      5,
+			QueryTimeout:                    30,
+			MaterializedViewRefreshInterval: 2, // Default: refresh every 2 minutes
 		},
 	},
 	Dispatcher: DispatcherConfiguration{
@@ -376,7 +378,9 @@ type MetricsConfiguration struct {
 }
 
 type PrometheusMetricsConfiguration struct {
-	SampleTime uint64 `json:"sample_time" envconfig:"CONVOY_METRICS_SAMPLE_TIME"`
+	SampleTime                      uint64 `json:"sample_time" envconfig:"CONVOY_METRICS_SAMPLE_TIME"`
+	QueryTimeout                    uint64 `json:"query_timeout" envconfig:"CONVOY_METRICS_QUERY_TIMEOUT"` // Timeout in seconds for metrics collection queries
+	MaterializedViewRefreshInterval uint64 `json:"materialized_view_refresh_interval" envconfig:"CONVOY_METRICS_MATERIALIZED_VIEW_REFRESH_INTERVAL"`
 }
 
 const (

config/config_test.go

@@ -179,7 +179,9 @@ func TestLoadConfig(t *testing.T) {
 					IsEnabled: false,
 					Backend:   "prometheus",
 					Prometheus: PrometheusMetricsConfiguration{
-						SampleTime: 5,
+						SampleTime:                      5,
+						QueryTimeout:                    30,
+						MaterializedViewRefreshInterval: 2,
 					},
 				},
 				Dispatcher: DispatcherConfiguration{
@@ -276,7 +278,9 @@ func TestLoadConfig(t *testing.T) {
 					IsEnabled: false,
 					Backend:   "prometheus",
 					Prometheus: PrometheusMetricsConfiguration{
-						SampleTime: 5,
+						SampleTime:                      5,
+						QueryTimeout:                    30,
+						MaterializedViewRefreshInterval: 2,
 					},
 				},
 				Dispatcher: DispatcherConfiguration{
@@ -372,7 +376,9 @@ func TestLoadConfig(t *testing.T) {
 					IsEnabled: false,
 					Backend:   "prometheus",
 					Prometheus: PrometheusMetricsConfiguration{
-						SampleTime: 5,
+						SampleTime:                      5,
+						QueryTimeout:                    30,
+						MaterializedViewRefreshInterval: 2,
 					},
 				},
 				Dispatcher: DispatcherConfiguration{

convoy.json.example

@@ -100,7 +100,9 @@
   "metrics": {
         "metrics_backend": "prometheus",
         "prometheus_metrics": {
-            "sample_time": 10
+            "sample_time": 5,
+            "query_timeout": 30,
+            "materialized_view_refresh_interval": 2
         }
   }
 }