chore(ses): deduplicate config load and update email

flow/alerting/alerting.go

@@ -114,7 +114,7 @@ func (a *Alerter) AlertIfSlotLag(ctx context.Context, peerName string, slotInfo
 		}
 	}
 
-	alertKey := peerName + "-slot-lag-threshold-exceeded"
+	alertKey := fmt.Sprintf("%s Slot Lag Threshold Exceeded for Peer %s", deploymentUIDPrefix, peerName)
 	alertMessageTemplate := fmt.Sprintf("%sSlot `%s` on peer `%s` has exceeded threshold size of %%dMB, "+
 		`currently at %.2fMB!`, deploymentUIDPrefix, slotInfo.SlotName, peerName, slotInfo.LagInMb)
 
@@ -147,7 +147,7 @@ func (a *Alerter) AlertIfOpenConnections(ctx context.Context, peerName string,
 
 	deploymentUIDPrefix := ""
 	if peerdbenv.PeerDBDeploymentUID() != "" {
-		deploymentUIDPrefix = fmt.Sprintf("[%s] ", peerdbenv.PeerDBDeploymentUID())
+		deploymentUIDPrefix = fmt.Sprintf("[%s] - ", peerdbenv.PeerDBDeploymentUID())
 	}
 
 	// same as with slot lag, use lowest threshold for catalog
@@ -159,7 +159,7 @@ func (a *Alerter) AlertIfOpenConnections(ctx context.Context, peerName string,
 		}
 	}
 
-	alertKey := peerName + "-max-open-connections-threshold-exceeded"
+	alertKey := fmt.Sprintf("%s Max Open Connections Threshold Exceeded for Peer %s", deploymentUIDPrefix, peerName)
 	alertMessageTemplate := fmt.Sprintf("%sOpen connections from PeerDB user `%s` on peer `%s`"+
 		` has exceeded threshold size of %%d connections, currently at %d connections!`,
 		deploymentUIDPrefix, openConnections.UserName, peerName, openConnections.CurrentOpenConnections)

flow/alerting/email_alert_sender.go

@@ -2,9 +2,9 @@ package alerting
 
 import (
 	"context"
+	"github.com/PeerDB-io/peer-flow/shared/aws_common"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/ses"
 	"github.com/aws/aws-sdk-go-v2/service/ses/types"
 
@@ -89,15 +89,10 @@ func NewEmailAlertSender(client *ses.Client, config *EmailAlertSenderConfig) Ema
 }
 
 func newSesClient(ctx context.Context, region *string) (*ses.Client, error) {
-	sdkConfig, err := config.LoadDefaultConfig(ctx, func(options *config.LoadOptions) error {
-		if region != nil {
-			options.Region = *region
-		}
-		return nil
-	})
+	sdkConfig, err := aws_common.LoadSdkConfig(ctx, region)
 	if err != nil {
 		return nil, err
 	}
-	snsClient := ses.NewFromConfig(sdkConfig)
+	snsClient := ses.NewFromConfig(*sdkConfig)
 	return snsClient, nil
 }

flow/shared/aws_common/config.go

@@ -0,0 +1,21 @@
+package aws_common
+
+import (
+	"context"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+)
+
+func LoadSdkConfig(ctx context.Context, region *string) (*aws.Config, error) {
+	sdkConfig, err := config.LoadDefaultConfig(ctx, func(options *config.LoadOptions) error {
+		if region != nil {
+			options.Region = *region
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &sdkConfig, nil
+
+}

flow/shared/telemetry/sns_message_sender.go

@@ -4,10 +4,10 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"github.com/PeerDB-io/peer-flow/shared/aws_common"
 	"strings"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/sns"
 	"github.com/aws/aws-sdk-go-v2/service/sns/types"
 	"go.temporal.io/sdk/activity"
@@ -96,15 +96,10 @@ func NewSNSMessageSender(client *sns.Client, config *SNSMessageSenderConfig) SNS
 }
 
 func newSnsClient(ctx context.Context, region *string) (*sns.Client, error) {
-	sdkConfig, err := config.LoadDefaultConfig(ctx, func(options *config.LoadOptions) error {
-		if region != nil {
-			options.Region = *region
-		}
-		return nil
-	})
+	sdkConfig, err := aws_common.LoadSdkConfig(ctx, region)
 	if err != nil {
 		return nil, err
 	}
-	snsClient := sns.NewFromConfig(sdkConfig)
+	snsClient := sns.NewFromConfig(*sdkConfig)
 	return snsClient, nil
 }