try to fix types

PeerDB-io · Mar 20, 2024 · b1b106f · b1b106f
1 parent fcd5f59
commit b1b106f
Showing 1 changed file with 39 additions and 11 deletions.
diff --git a/flow/dynamicconf/dynamicconf.go b/flow/dynamicconf/dynamicconf.go
@@ -25,7 +25,35 @@ func dynamicConfKeyExists(ctx context.Context, conn *pgxpool.Pool, key string) b
 	return exists.Bool
 }
 
-func dynamicConfNumber[T constraints.Integer](ctx context.Context, key string, defaultValue T) T {
+func dynamicConfSigned[T constraints.Signed](ctx context.Context, key string, defaultValue T) T {
+	conn, err := utils.GetCatalogConnectionPoolFromEnv(ctx)
+	if err != nil {
+		logger.LoggerFromCtx(ctx).Error("Failed to get catalog connection pool: %v", err)
+		return defaultValue
+	}
+
+	if !dynamicConfKeyExists(ctx, conn, key) {
+		return defaultValue
+	}
+
+	var value pgtype.Text
+	query := "SELECT config_value FROM alerting_settings WHERE config_name = $1"
+	err = conn.QueryRow(ctx, query, key).Scan(&value)
+	if err != nil {
+		logger.LoggerFromCtx(ctx).Error("Failed to get key: %v", err)
+		return defaultValue
+	}
+
+	result, err := strconv.ParseInt(value.String, 10, 32)
+	if err != nil {
+		logger.LoggerFromCtx(ctx).Error("Failed to parse uint32: %v", err)
+		return defaultValue
+	}
+
+	return T(result)
+}
+
+func dynamicConfUnsigned[T constraints.Unsigned](ctx context.Context, key string, defaultValue T) T {
 	conn, err := utils.GetCatalogConnectionPoolFromEnv(ctx)
 	if err != nil {
 		logger.LoggerFromCtx(ctx).Error("Failed to get catalog connection pool: %v", err)
@@ -55,46 +83,46 @@ func dynamicConfNumber[T constraints.Integer](ctx context.Context, key string, d
 
 // PEERDB_SLOT_LAG_MB_ALERT_THRESHOLD, 0 disables slot lag alerting entirely
 func PeerDBSlotLagMBAlertThreshold(ctx context.Context) uint32 {
-	return dynamicConfNumber[uint32](ctx, "PEERDB_SLOT_LAG_MB_ALERT_THRESHOLD", 5000)
+	return dynamicConfUnsigned[uint32](ctx, "PEERDB_SLOT_LAG_MB_ALERT_THRESHOLD", 5000)
 }
 
 // PEERDB_ALERTING_GAP_MINUTES, 0 disables all alerting entirely
 func PeerDBAlertingGapMinutesAsDuration(ctx context.Context) time.Duration {
-	why := dynamicConfNumber[uint32](ctx, "PEERDB_ALERTING_GAP_MINUTES", 15)
+	why := dynamicConfUnsigned[uint32](ctx, "PEERDB_ALERTING_GAP_MINUTES", 15)
 	return time.Duration(why) * time.Minute
 }
 
 // PEERDB_PGPEER_OPEN_CONNECTIONS_ALERT_THRESHOLD, 0 disables open connections alerting entirely
 func PeerDBOpenConnectionsAlertThreshold(ctx context.Context) uint32 {
-	return dynamicConfNumber[uint32](ctx, "PEERDB_PGPEER_OPEN_CONNECTIONS_ALERT_THRESHOLD", 5)
+	return dynamicConfUnsigned[uint32](ctx, "PEERDB_PGPEER_OPEN_CONNECTIONS_ALERT_THRESHOLD", 5)
 }
 
 func PeerDBSnowflakeMergeParallelism(ctx context.Context) int {
-	return int(dynamicConfNumber[int32](ctx, "PEERDB_SNOWFLAKE_MERGE_PARALLELISM", 8))
+	return dynamicConfSigned(ctx, "PEERDB_SNOWFLAKE_MERGE_PARALLELISM", 8)
 }
 
 // PEERDB_CDC_DISK_SPILL_RECORDS_THRESHOLD
 func PeerDBCDCDiskSpillRecordsThreshold(ctx context.Context) int64 {
-	return dynamicConfNumber[int64](ctx, "PEERDB_CDC_DISK_SPILL_RECORDS_THRESHOLD", 1_000_000)
+	return dynamicConfSigned[int64](ctx, "PEERDB_CDC_DISK_SPILL_RECORDS_THRESHOLD", 1_000_000)
 }
 
 // PEERDB_CDC_DISK_SPILL_RECORDS_THRESHOLD, negative numbers means memory threshold disabled
 func PeerDBCDCDiskSpillMemPercentThreshold(ctx context.Context) int64 {
-	return dynamicConfNumber[int64](ctx, "PEERDB_CDC_DISK_SPILL_MEM_PERCENT_THRESHOLD", -1)
+	return dynamicConfSigned[int64](ctx, "PEERDB_CDC_DISK_SPILL_MEM_PERCENT_THRESHOLD", -1)
 }
 
 // PEERDB_CDC_CHANNEL_BUFFER_SIZE
 func PeerDBCDCChannelBufferSize(ctx context.Context) int {
-	return int(dynamicConfNumber[int32](ctx, "PEERDB_CDC_CHANNEL_BUFFER_SIZE", 1<<18))
+	return dynamicConfSigned(ctx, "PEERDB_CDC_CHANNEL_BUFFER_SIZE", 1<<18)
 }
 
 // PEERDB_EVENTHUB_FLUSH_TIMEOUT_SECONDS
 func PeerDBEventhubFlushTimeoutSeconds(ctx context.Context) time.Duration {
-	x := dynamicConfNumber[int64](ctx, "PEERDB_EVENTHUB_FLUSH_TIMEOUT_SECONDS", 10)
+	x := dynamicConfSigned[int64](ctx, "PEERDB_EVENTHUB_FLUSH_TIMEOUT_SECONDS", 10)
 	return time.Duration(x) * time.Second
 }
 
 // GOMEMLIMIT is a variable internal to Golang itself, we use this for internal targets, 0 means no maximum
-func PeerDBFlowWorkerMaxMemBytes(ctx context.Context) uint64 {
-	return dynamicConfNumber[uint64](ctx, "GOMEMLIMIT", 0)
+func PeerDBFlowWorkerMaxMemBytes(ctx context.Context) uint32 {
+	return dynamicConfUnsigned[uint32](ctx, "GOMEMLIMIT", 0)
 }