removed ENABLE_STATS option, checking for catalog connectivity and co…

…de cleanup

docker-compose.yml

@@ -22,8 +22,6 @@ x-flow-worker-env: &flow-worker-env
   ENABLE_PROFILING: "true"
   # enables exporting of mirror metrics to Prometheus for visualization using Grafana
   ENABLE_METRICS: "true"
-  # enables exporting of mirror metrics to Catalog in the PEERDB_STATS schema.
-  ENABLE_STATS: "true"
   PYROSCOPE_SERVER_ADDRESS: http://pyroscope:4040
 
 services:

flow/cmd/main.go

@@ -44,13 +44,6 @@ func main() {
 		EnvVars: []string{"ENABLE_METRICS"},
 	}
 
-	monitoringFlag := &cli.BoolFlag{
-		Name:    "enable-monitoring",
-		Value:   false, // Default is off
-		Usage:   "Enable mirror monitoring for the application",
-		EnvVars: []string{"ENABLE_STATS"},
-	}
-
 	pyroscopeServerFlag := &cli.StringFlag{
 		Name:    "pyroscope-server-address",
 		Value:   "http://pyroscope:4040",
@@ -76,7 +69,6 @@ func main() {
 						TemporalHostPort: temporalHostPort,
 						EnableProfiling:  ctx.Bool("enable-profiling"),
 						EnableMetrics:    ctx.Bool("enable-metrics"),
-						EnableMonitoring: ctx.Bool("enable-monitoring"),
 						PyroscopeServer:  ctx.String("pyroscope-server-address"),
 						MetricsServer:    ctx.String("metrics-server"),
 					})
@@ -85,7 +77,6 @@ func main() {
 					temporalHostPortFlag,
 					profilingFlag,
 					metricsFlag,
-					monitoringFlag,
 					pyroscopeServerFlag,
 					metricsServerFlag,
 				},

flow/cmd/worker.go

@@ -8,9 +8,6 @@ import (
 	"syscall"
 	"time"
 
-	//nolint:gosec
-	_ "net/http/pprof"
-
 	"github.com/PeerDB-io/peer-flow/activities"
 	utils "github.com/PeerDB-io/peer-flow/connectors/utils/catalog"
 	"github.com/PeerDB-io/peer-flow/connectors/utils/monitoring"
@@ -31,7 +28,6 @@ type WorkerOptions struct {
 	TemporalHostPort string
 	EnableProfiling  bool
 	EnableMetrics    bool
-	EnableMonitoring bool
 	PyroscopeServer  string
 	MetricsServer    string
 }
@@ -111,14 +107,11 @@ func WorkerMain(opts *WorkerOptions) error {
 		}
 	}
 
-	catalogMirrorMonitor := monitoring.NewCatalogMirrorMonitor(nil)
-	if opts.EnableMonitoring {
-		conn, err := utils.GetCatalogConnectionPoolFromEnv()
-		if err != nil {
-			return fmt.Errorf("unable to create catalog connection pool: %w", err)
-		}
-		catalogMirrorMonitor = monitoring.NewCatalogMirrorMonitor(conn)
+	conn, err := utils.GetCatalogConnectionPoolFromEnv()
+	if err != nil {
+		return fmt.Errorf("unable to create catalog connection pool: %w", err)
 	}
+	catalogMirrorMonitor := monitoring.NewCatalogMirrorMonitor(conn)
 	defer catalogMirrorMonitor.Close()
 
 	c, err := client.Dial(clientOptions)

flow/connectors/utils/catalog/env.go

@@ -22,6 +22,11 @@ func GetCatalogConnectionPoolFromEnv() (*pgxpool.Pool, error) {
 		return nil, fmt.Errorf("unable to establish connection with catalog: %w", err)
 	}
 
+	err = catalogConn.Ping(context.Background())
+	if err != nil {
+		return nil, fmt.Errorf("unable to establish connection with catalog: %w", err)
+	}
+
 	return catalogConn, nil
 }