refactor: use different env and dedup strategy

PeerDB-io · Mar 4, 2024 · 1286635 · 1286635
1 parent 0214642
commit 1286635
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 6 deletions.
diff --git a/flow/alerting/alerting.go b/flow/alerting/alerting.go
@@ -4,13 +4,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"go.temporal.io/sdk/activity"
 	"log/slog"
-	"os"
 	"time"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+	"go.temporal.io/sdk/activity"
 
 	"github.com/PeerDB-io/peer-flow/dynamicconf"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
@@ -58,7 +57,7 @@ func NewAlerter(ctx context.Context, catalogPool *pgxpool.Pool) *Alerter {
 	if catalogPool == nil {
 		panic("catalog pool is nil for Alerter")
 	}
-	snsTopic := os.Getenv("TELEMETRY_AWS_SNS_TOPIC_ARN")
+	snsTopic := peerdbenv.PeerDBTelemetryAWSSNSTopicArn()
 	var snsMessageSender telemetry.Sender
 	if snsTopic != "" {
 		var err error

diff --git a/flow/peerdbenv/config.go b/flow/peerdbenv/config.go
@@ -90,3 +90,8 @@ func PeerDBEnableWALHeartbeat() bool {
 func PeerDBEnableParallelSyncNormalize() bool {
 	return getEnvBool("PEERDB_ENABLE_PARALLEL_SYNC_NORMALIZE", false)
 }
+
+// PEERDB_TELEMETRY_AWS_SNS_TOPIC_ARN
+func PeerDBTelemetryAWSSNSTopicArn() string {
+	return getEnvString("PEERDB_TELEMETRY_AWS_SNS_TOPIC_ARN", "")
+}
diff --git a/flow/shared/telemetry/sns_message_sender.go b/flow/shared/telemetry/sns_message_sender.go
@@ -4,14 +4,13 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
-	"fmt"
 	"strings"
-	"time"
 
 	"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"
 )
 
 type SNSMessageSender interface {
@@ -28,7 +27,13 @@ type SNSMessageSenderConfig struct {
 }
 
 func (s *SNSMessageSenderImpl) SendMessage(ctx context.Context, subject string, body string, attributes Attributes) (*string, error) {
-	deduplicationString := fmt.Sprintf("[%s] - [%s] - [Window: %s]", attributes.DeploymentUID, subject, time.Now().Truncate(30*time.Minute))
+	activityInfo := activity.GetInfo(ctx)
+	deduplicationString := strings.Join([]string{
+		"deployID", attributes.DeploymentUID,
+		"subject", subject,
+		"runID", activityInfo.WorkflowExecution.RunID,
+		"activityName", activityInfo.ActivityType.Name,
+	}, " || ")
 	h := sha256.New()
 	h.Write([]byte(deduplicationString))
 	deduplicationHash := hex.EncodeToString(h.Sum(nil))