From 565d54e1eeb4223b2e03fdcb3ca1404dd9f64b58 Mon Sep 17 00:00:00 2001
From: Kaushik Iska <iska.kaushik@gmail.com>
Date: Thu, 2 Jan 2025 11:06:56 -0600
Subject: [PATCH] Stop sending non-actionable errors to incident.io (#2397)

---
 flow/alerting/alerting.go                     |  8 +++----
 flow/shared/err_types.go                      | 24 +++++++++++++++++++
 .../telemetry/incidentio_message_sender.go    | 12 ++++++++++
 3 files changed, 40 insertions(+), 4 deletions(-)
 create mode 100644 flow/shared/err_types.go

diff --git a/flow/alerting/alerting.go b/flow/alerting/alerting.go
index d1394561fd..fa1ebf21f9 100644
--- a/flow/alerting/alerting.go
+++ b/flow/alerting/alerting.go
@@ -440,13 +440,13 @@ func (a *Alerter) LogFlowError(ctx context.Context, flowName string, err error)
 	}
 	var tags []string
 	if errors.Is(err, context.Canceled) {
-		tags = append(tags, "err:Canceled")
+		tags = append(tags, string(shared.ErrTypeCanceled))
 	}
 	if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
-		tags = append(tags, "err:EOF")
+		tags = append(tags, string(shared.ErrTypeEOF))
 	}
 	if errors.Is(err, net.ErrClosed) {
-		tags = append(tags, "err:Closed")
+		tags = append(tags, string(shared.ErrTypeClosed))
 	}
 	var pgErr *pgconn.PgError
 	if errors.As(err, &pgErr) {
@@ -454,7 +454,7 @@ func (a *Alerter) LogFlowError(ctx context.Context, flowName string, err error)
 	}
 	var netErr *net.OpError
 	if errors.As(err, &netErr) {
-		tags = append(tags, "err:Net")
+		tags = append(tags, string(shared.ErrTypeNet))
 	}
 	a.sendTelemetryMessage(ctx, logger, flowName, errorWithStack, telemetry.ERROR, tags...)
 }
diff --git a/flow/shared/err_types.go b/flow/shared/err_types.go
new file mode 100644
index 0000000000..171b11ca6c
--- /dev/null
+++ b/flow/shared/err_types.go
@@ -0,0 +1,24 @@
+package shared
+
+type ErrType string
+
+const (
+	ErrTypeCanceled ErrType = "err:Canceled"
+	ErrTypeClosed   ErrType = "err:Closed"
+	ErrTypeNet      ErrType = "err:Net"
+	ErrTypeEOF      ErrType = "err:EOF"
+)
+
+func SkipSendingToIncidentIo(errTags []string) bool {
+	skipTags := map[string]struct{}{
+		string(ErrTypeCanceled): {},
+		string(ErrTypeClosed):   {},
+		string(ErrTypeNet):      {},
+	}
+	for _, tag := range errTags {
+		if _, ok := skipTags[tag]; ok {
+			return true
+		}
+	}
+	return false
+}
diff --git a/flow/shared/telemetry/incidentio_message_sender.go b/flow/shared/telemetry/incidentio_message_sender.go
index ead24dd9a6..dbcbf80b0d 100644
--- a/flow/shared/telemetry/incidentio_message_sender.go
+++ b/flow/shared/telemetry/incidentio_message_sender.go
@@ -8,11 +8,14 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
 	"strings"
 	"time"
 
 	"go.temporal.io/sdk/activity"
+
+	"github.com/PeerDB-io/peer-flow/shared"
 )
 
 type IncidentIoAlert struct {
@@ -50,6 +53,15 @@ func (i *IncidentIoMessageSender) SendMessage(
 		activityInfo = activity.GetInfo(ctx)
 	}
 
+	if shared.SkipSendingToIncidentIo(attributes.Tags) {
+		logger := shared.LoggerFromCtx(ctx)
+		logger.Info("skipping incident.io alert",
+			slog.Any("attributes", attributes),
+			slog.String("subject", subject),
+			slog.String("body", body))
+		return "", nil
+	}
+
 	deduplicationString := strings.Join([]string{
 		"deployID", attributes.DeploymentUID,
 		"subject", subject,