Remove SNS and telemetry sender infra (#4492)

Confirmed not used outside decommissioned Cloud

Author: ilidemi

codecov.yml

@@ -15,7 +15,6 @@ ignore:
   - "flow/e2e/eventhub"
   - "flow/middleware"
   - "flow/shared/aws_common"
-  - "flow/shared/telemetry/sns_message_sender.go"
 
 component_management:
   individual_components:

flow/activities/maintenance_activity.go

@@ -25,7 +25,6 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	"github.com/PeerDB-io/peerdb/flow/shared/telemetry"
 )
 
 const (
@@ -382,7 +381,6 @@ func (a *MaintenanceActivity) BackgroundAlerter(ctx context.Context) error {
 			activity.RecordHeartbeat(ctx, "Maintenance Workflow is still running")
 		case <-alertTicker.C:
 			slog.WarnContext(ctx, "Maintenance Workflow is still running")
-			a.Alerter.EmitNonFlowWarningTelemetryEvent(ctx, telemetry.MaintenanceWait, "Waiting", "Maintenance mode is still running")
 			a.OtelManager.Metrics.MaintenanceStatusGauge.Record(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
 				attribute.String(otel_metrics.WorkflowTypeKey, activity.GetInfo(ctx).WorkflowType.Name),
 			)))

flow/activities/snapshot_activity.go

@@ -54,7 +54,6 @@ func (a *SnapshotActivity) CloseSlotKeepAlive(ctx context.Context, flowJobName s
 		s.connectorClose(ctx)
 		delete(a.SlotSnapshotStates, flowJobName)
 	}
-	a.Alerter.EmitFlowInfoTelemetryEvent(ctx, flowJobName, "Ended Snapshot Flow Job")
 
 	return nil
 }
@@ -66,7 +65,6 @@ func (a *SnapshotActivity) SetupReplication(
 	ctx = context.WithValue(ctx, shared.FlowNameKey, config.FlowJobName)
 	logger := internal.LoggerFromCtx(ctx)
 	a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "Setting up replication slot and publication")
-	a.Alerter.EmitFlowInfoTelemetryEvent(ctx, config.FlowJobName, "Started Snapshot Flow Job")
 
 	conn, connClose, err := connectors.GetByNameAs[connectors.CDCPullConnectorCore](ctx, nil, a.CatalogPool, config.PeerName)
 	if err != nil {

flow/alerting/alerting.go

@@ -5,34 +5,25 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"log/slog"
-	"net"
 	"slices"
 	"strings"
 	"time"
 
-	"github.com/ClickHouse/clickhouse-go/v2"
-	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/jackc/pgx/v5"
-	"github.com/jackc/pgx/v5/pgconn"
-	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
-	"golang.org/x/crypto/ssh"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	"github.com/PeerDB-io/peerdb/flow/shared/telemetry"
 )
 
 // alerting service, no cool name :(
 type Alerter struct {
 	shared.CatalogPool
-	snsTelemetrySender telemetry.Sender
-	otelManager        *otel_metrics.OtelManager
+	otelManager *otel_metrics.OtelManager
 }
 
 type AlertSenderConfig struct {
@@ -48,27 +39,13 @@ type AlertKeys struct {
 }
 
 // doesn't take care of closing pool, needs to be done externally.
-func NewAlerter(ctx context.Context, catalogPool shared.CatalogPool, otelManager *otel_metrics.OtelManager) *Alerter {
+func NewAlerter(_ context.Context, catalogPool shared.CatalogPool, otelManager *otel_metrics.OtelManager) *Alerter {
 	if catalogPool.Pool == nil {
 		panic("catalog pool is nil for Alerter")
 	}
-	snsTopic := internal.PeerDBTelemetryAWSSNSTopicArn()
-	var snsMessageSender telemetry.Sender
-	if snsTopic != "" {
-		var err error
-		snsMessageSender, err = telemetry.NewSNSMessageSenderWithNewClient(ctx, &telemetry.SNSMessageSenderConfig{
-			Topic: snsTopic,
-		})
-		internal.LoggerFromCtx(ctx).Info("Successfully registered sns telemetry sender")
-		if err != nil {
-			panic(fmt.Sprintf("unable to setup telemetry is nil for Alerter %+v", err))
-		}
-	}
-
 	return &Alerter{
-		CatalogPool:        catalogPool,
-		snsTelemetrySender: snsMessageSender,
-		otelManager:        otelManager,
+		CatalogPool: catalogPool,
+		otelManager: otelManager,
 	}
 }
 
@@ -350,71 +327,6 @@ func (a *Alerter) checkAndAddAlertToCatalog(ctx context.Context, alertConfigId i
 	return false
 }
 
-func (a *Alerter) sendTelemetryMessage(
-	ctx context.Context,
-	flowName string,
-	more string,
-	level telemetry.Level,
-	additionalTags ...string,
-) {
-	logger := internal.LoggerFromCtx(ctx)
-	allTags := []string{flowName, internal.PeerDBDeploymentUID()}
-	allTags = append(allTags, additionalTags...)
-
-	if flowTags, err := GetTags(ctx, a.CatalogPool, flowName); err != nil {
-		logger.Warn("failed to get flow tags", slog.Any("error", err))
-	} else {
-		for key, value := range flowTags {
-			allTags = append(allTags, fmt.Sprintf("%s:%s", key, value))
-		}
-	}
-
-	details := fmt.Sprintf("[%s] %s", flowName, more)
-	attributes := telemetry.Attributes{
-		Level:         level,
-		DeploymentUID: internal.PeerDBDeploymentUID(),
-		Tags:          allTags,
-		Type:          flowName,
-	}
-
-	if a.snsTelemetrySender != nil {
-		if response, err := a.snsTelemetrySender.SendMessage(ctx, details, details, attributes); err != nil {
-			logger.Warn("failed to send message to snsTelemetrySender", slog.Any("error", err))
-		} else {
-			logger.Debug("received response from snsTelemetrySender", slog.String("response", response))
-		}
-	}
-}
-
-func (a *Alerter) emitNonFlowTelemetryEvent(
-	ctx context.Context, eventType telemetry.EventType, key string, message string, level telemetry.Level,
-) {
-	a.sendTelemetryMessage(ctx, string(eventType)+":"+key, message, level)
-}
-
-// Wrapper for different telemetry levels for non-flow events
-func (a *Alerter) EmitNonFlowInfoTelemetryEvent(ctx context.Context, eventType telemetry.EventType, key string, message string) {
-	a.emitNonFlowTelemetryEvent(ctx, eventType, key, message, telemetry.INFO)
-}
-
-func (a *Alerter) EmitNonFlowWarningTelemetryEvent(ctx context.Context, eventType telemetry.EventType, key string, message string) {
-	a.emitNonFlowTelemetryEvent(ctx, eventType, key, message, telemetry.WARN)
-}
-
-func (a *Alerter) EmitNonFlowErrorTelemetryEvent(ctx context.Context, eventType telemetry.EventType, key string, message string) {
-	a.emitNonFlowTelemetryEvent(ctx, eventType, key, message, telemetry.ERROR)
-}
-
-func (a *Alerter) EmitNonFlowCriticalTelemetryEvent(ctx context.Context, eventType telemetry.EventType, key string, message string) {
-	a.emitNonFlowTelemetryEvent(ctx, eventType, key, message, telemetry.CRITICAL)
-}
-
-func (a *Alerter) EmitFlowInfoTelemetryEvent(ctx context.Context, flowName string, info string) {
-	logger := internal.LoggerFromCtx(ctx)
-	logger.Info(info)
-	a.sendTelemetryMessage(ctx, flowName, info, telemetry.INFO)
-}
-
 type FlowErrorType string
 
 func (f FlowErrorType) String() string {
@@ -427,7 +339,7 @@ const (
 	FlowErrorTypeError FlowErrorType = "error"
 )
 
-// recordFlowErrorInternal pushes the error to the errors table and emits a metric as well as a telemetry message.
+// recordFlowErrorInternal pushes the error to the errors table and emits a metric.
 func (a *Alerter) recordFlowErrorInternal(
 	ctx context.Context,
 	flowName string,
@@ -460,42 +372,7 @@ func (a *Alerter) recordFlowErrorInternal(
 		logger.Error("failed to insert flow error", slog.Any("error", err))
 	}
 
-	// 3. Only send alerts to telemetry sender if the env is enabled
-	if internal.PeerDBTelemetrySenderSendErrorAlertsEnabled(ctx) {
-		var tags []string
-		if errors.Is(err, context.Canceled) {
-			tags = append(tags, string(shared.ErrTypeCanceled))
-		}
-		if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
-			tags = append(tags, string(shared.ErrTypeEOF))
-		}
-		if errors.Is(err, net.ErrClosed) || strings.HasSuffix(err.Error(), "use of closed network connection") {
-			tags = append(tags, string(shared.ErrTypeClosed))
-		}
-		if pgErr, ok := errors.AsType[*pgconn.PgError](err); ok {
-			tags = append(tags, "pgcode:"+pgErr.Code)
-		}
-		if myErr, ok := errors.AsType[*mysql.MyError](err); ok {
-			tags = append(tags, fmt.Sprintf("mycode:%d", myErr.Code), "mystate:"+myErr.State)
-		}
-		if mongoErr, ok := errors.AsType[*driver.Error](err); ok {
-			tags = append(tags, fmt.Sprintf("mongocode:%d", mongoErr.Code))
-		}
-		if chErr, ok := errors.AsType[*clickhouse.Exception](err); ok {
-			tags = append(tags, fmt.Sprintf("chcode:%d", chErr.Code))
-		}
-		if _, ok := errors.AsType[*net.OpError](err); ok {
-			tags = append(tags, string(shared.ErrTypeNet))
-		}
-		// For SSH connection errors, we currently tag them as "err:Net"
-		if _, ok := errors.AsType[*ssh.OpenChannelError](err); ok {
-			tags = append(tags, string(shared.ErrTypeNet))
-		}
-		tags = append(tags, "errorClass:"+errClass.String(), "errorAction:"+errClass.ErrorAction().String())
-		a.sendTelemetryMessage(ctx, flowName, errMessage, telemetry.ERROR, tags...)
-	}
-
-	// 4. Record error metrics
+	// 3. Record error metrics
 	errorAttributes := []attribute.KeyValue{
 		attribute.Stringer(otel_metrics.ErrorClassKey, errClass),
 		attribute.Stringer(otel_metrics.ErrorActionKey, errClass.ErrorAction()),

flow/go.mod

@@ -30,7 +30,6 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/kms v1.52.0
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.101.0
 	github.com/aws/aws-sdk-go-v2/service/ses v1.34.24
-	github.com/aws/aws-sdk-go-v2/service/sns v1.39.17
 	github.com/aws/aws-sdk-go-v2/service/sts v1.42.1
 	github.com/aws/smithy-go v1.25.1
 	github.com/cockroachdb/pebble/v2 v2.1.5

flow/go.sum

@@ -163,8 +163,6 @@ github.com/aws/aws-sdk-go-v2/service/ses v1.34.24 h1:ZPCoU087iv2IWJKNQyN+MYOqZAM
 github.com/aws/aws-sdk-go-v2/service/ses v1.34.24/go.mod h1:4T8OWyQ9nkdUrcqHHJkyibP5TfcWDTe4kWL40tiCOcs=
 github.com/aws/aws-sdk-go-v2/service/signin v1.0.11 h1:TdJ+HdzOBhU8+iVAOGUTU63VXopcumCOF1paFulHWZc=
 github.com/aws/aws-sdk-go-v2/service/signin v1.0.11/go.mod h1:R82ZRExE/nheo0N+T8zHPcLRTcH8MGsnR3BiVGX0TwI=
-github.com/aws/aws-sdk-go-v2/service/sns v1.39.17 h1:synXIPC/L4Cc489P0XDcrVJzHSLj7krKRpFLalbGM2k=
-github.com/aws/aws-sdk-go-v2/service/sns v1.39.17/go.mod h1:4ABZnI23uNK37waIjGwkubnCwGhepIt9x1GvASfljJA=
 github.com/aws/aws-sdk-go-v2/service/sso v1.30.17 h1:7byT8HUWrgoRp6sXjxtZwgOKfhss5fW6SkLBtqzgRoE=
 github.com/aws/aws-sdk-go-v2/service/sso v1.30.17/go.mod h1:xNWknVi4Ezm1vg1QsB/5EWpAJURq22uqd38U8qKvOJc=
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.0 h1:nDARhv/oF55bcxF7rCI/4PDxOKnVXVWwDuDwCs2I2SQ=

flow/internal/config.go

@@ -78,11 +78,6 @@ func PeerDBCatalogRequireTls() bool {
 	return getEnvConvert("PEERDB_CATALOG_REQUIRE_TLS", false, strconv.ParseBool)
 }
 
-// PEERDB_TELEMETRY_AWS_SNS_TOPIC_ARN
-func PeerDBTelemetryAWSSNSTopicArn() string {
-	return GetEnvString("PEERDB_TELEMETRY_AWS_SNS_TOPIC_ARN", "")
-}
-
 func PeerDBAlertingEmailSenderSourceEmail() string {
 	return GetEnvString("PEERDB_ALERTING_EMAIL_SENDER_SOURCE_EMAIL", "")
 }
@@ -195,16 +190,6 @@ func PeerDBSetupFlowWorkflowTaskTimeoutSeconds() int {
 	return getEnvConvert("PEERDB_SETUP_FLOW_WORKFLOW_TASK_TIMEOUT_SECONDS", 60, strconv.Atoi)
 }
 
-// PEERDB_TELEMETRY_SENDER_SEND_ERROR_ALERTS_ENABLED is whether to send error alerts to the telemetry sender
-func PeerDBTelemetrySenderSendErrorAlertsEnabled(ctx context.Context) bool {
-	enabled, err := strconv.ParseBool(GetEnvString("PEERDB_TELEMETRY_SENDER_SEND_ERROR_ALERTS_ENABLED", "false"))
-	if err != nil {
-		slog.ErrorContext(ctx, "failed to parse PEERDB_TELEMETRY_SENDER_SEND_ERROR_ALERTS_ENABLED to bool", slog.Any("error", err))
-		return false
-	}
-	return enabled
-}
-
 // PEERDB_SWITCHBOARD_ENABLED enables the Switchboard server
 func PeerDBSwitchboardEnabled() bool {
 	return GetEnvBool("PEERDB_SWITCHBOARD_ENABLED", false)