Publish SNS event on new alert (#200)

Author: adrain-cb

.github/workflows/hygeine.yml

@@ -48,8 +48,8 @@ jobs:
     - name: Set up Node.js
       uses: actions/setup-node@v2
       with:
-        node-version: '20' 
-
+        node-version: '20'
+        
     - name: Install markdownlint CLI
       run: npm install -g markdownlint-cli
 

Makefile

@@ -27,11 +27,13 @@ go-gen-mocks:
 
 .PHONY: test
 test:
-	@ go test ./internal/... -timeout $(TEST_LIMIT)
+	@go test ./internal/... -timeout $(TEST_LIMIT)
 
 .PHONY: test-e2e
 e2e-test:
-	@ go test ./e2e/...  -timeout $(TEST_LIMIT) -deploy-config ../.devnet/devnetL1.json -parallel=4 -v
+	@docker compose up -d
+	@go test ./e2e/...  -timeout $(TEST_LIMIT) -deploy-config ../.devnet/devnetL1.json -parallel=4 -v
+	@docker compose down
 
 .PHONY: lint
 lint:

config.env.template

@@ -32,6 +32,9 @@ P0_PAGERDUTY_ALERT_EVENTS_URL=
 P1_PAGERDUTY_INTEGRATION_KEY=
 P1_PAGERDUTY_ALERT_EVENTS_URL=
 
+SNS_TOPIC_ARN=
+AWS_ENDPOINT=
+
 # Metrics configurations
 METRICS_HOST=localhost
 METRICS_PORT=7300

docker-compose.yml

@@ -0,0 +1,15 @@
+version: "3.8"
+
+services:
+  localstack:
+    container_name: "${LOCALSTACK_DOCKER_NAME:-localstack-main}"
+    image: localstack/localstack:3.1.0
+    ports:
+      - "127.0.0.1:4566:4566"            # LocalStack Gateway
+      - "127.0.0.1:4510-4559:4510-4559"  # external services port range
+    environment:
+      # LocalStack configuration: https://docs.localstack.cloud/references/configuration/
+      - DEBUG=${DEBUG:-0}
+    volumes:
+      - "/var/run/docker.sock:/var/run/docker.sock"
+      - "./scripts/localstack-e2e-test-setup.sh:/etc/localstack/init/ready.d/script.sh"

docs/alert-routing.md

@@ -32,10 +32,11 @@ a few examples on how you might want to configure your alert routing.
 
 Pessimism currently supports the following alert destinations:
 
-| Name      | Description                         |
-|-----------|-------------------------------------|
-| slack     | Sends alerts to a Slack channel     |
-| pagerduty | Sends alerts to a PagerDuty service |
+| Name      | Description                                       |
+|-----------|---------------------------------------------------|
+| slack     | Sends alerts to a Slack channel                   |
+| pagerduty | Sends alerts to a PagerDuty service               |
+| sns       | Sends alerts to an SNS topic defined in .env file |
 
 ## Alert Severity
 
@@ -47,6 +48,16 @@ Pessimism currently defines the following severities for alerts:
 | medium   | Alerts that could be hazardous, but may not be completely destructive       |
 | high     | Alerts that require immediate attention and could result in a loss of funds |
 
+## Publishing to an SNS Topic
+
+To publish alerts to an SNS topic, you must first create an SNS topic in the AWS
+console. Once you have created the topic, you will need to add the ARN of the
+topic to the `.env` file. Ensure that you have AWS_REGION,
+`AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` set in your environment if you are looking to publish messages to an SNS
+topic. The ARN should be added to the `SNS_TOPIC_ARN` variable found in the `.env` file.
+The AWS_ENDPOINT is optional and is primarily used for testing with localstack.
+> Note: Currently, Pessimism only support one SNS topic to publish alerts to.
+
 ## PagerDuty Severity Mapping
 
 PagerDuty supports the following severities: `critical`, `error`, `warning`,

e2e/alerting_test.go

@@ -3,6 +3,7 @@ package e2e_test
 import (
 	"context"
 	"math/big"
+
 	"testing"
 	"time"
 
@@ -17,11 +18,18 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+const (
+	// These are localstack specific Topic ARNs that are used to test the SNS integration.
+	MultiDirectiveTopicArn = "arn:aws:sns:us-east-1:000000000000:multi-directive-test-topic"
+	CoolDownTopicArn       = "arn:aws:sns:us-east-1:000000000000:alert-cooldown-test-topic"
+)
+
 // TestMultiDirectiveRouting ... Tests the E2E flow of a contract event heuristic with high priority alerts all
 // necessary destinations
 func TestMultiDirectiveRouting(t *testing.T) {
 
-	ts := e2e.CreateSysTestSuite(t)
+	ts := e2e.CreateSysTestSuite(t, MultiDirectiveTopicArn)
+	defer ts.Close()
 
 	updateSig := "ConfigUpdate(uint256,uint8,bytes)"
 	alertMsg := "System config gas config updated"
@@ -73,6 +81,12 @@ func TestMultiDirectiveRouting(t *testing.T) {
 		return height != nil && height.Uint64() > receipt.BlockNumber.Uint64(), nil
 	}))
 
+	snsMessages, err := e2e.GetSNSMessages(ts.AppCfg.AlertConfig.SNSConfig.Endpoint, "multi-directive-test-queue")
+	require.NoError(t, err)
+
+	assert.Len(t, snsMessages.Messages, 1, "Incorrect number of SNS messages sent")
+	assert.Contains(t, *snsMessages.Messages[0].Body, "contract_event", "System contract event alert was not sent")
+
 	slackPosts := ts.TestSlackSvr.SlackAlerts()
 	pdPosts := ts.TestPagerDutyServer.PagerDutyAlerts()
 
@@ -90,7 +104,7 @@ func TestMultiDirectiveRouting(t *testing.T) {
 // balance enforcement heuristic session on L2 network with a cooldown.
 func TestCoolDown(t *testing.T) {
 
-	ts := e2e.CreateSysTestSuite(t)
+	ts := e2e.CreateSysTestSuite(t, CoolDownTopicArn)
 	defer ts.Close()
 
 	alice := ts.Cfg.Secrets.Addresses().Alice
@@ -149,7 +163,7 @@ func TestCoolDown(t *testing.T) {
 	receipt, err := wait.ForReceipt(context.Background(), ts.L2Client, drainAliceTx.Hash(), types.ReceiptStatusSuccessful)
 	require.NoError(t, err)
 
-	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
+	require.NoError(t, wait.For(context.Background(), 1000*time.Millisecond, func() (bool, error) {
 		id := ids[0].PathID
 		height, err := ts.Subsystems.PathHeight(id)
 		if err != nil {
@@ -162,6 +176,11 @@ func TestCoolDown(t *testing.T) {
 	// Check that the balance enforcement was triggered using the mocked server cache.
 	posts := ts.TestSlackSvr.SlackAlerts()
 
+	sqsMessages, err := e2e.GetSNSMessages(ts.AppCfg.AlertConfig.SNSConfig.Endpoint, "alert-cooldown-test-queue")
+	assert.NoError(t, err)
+	assert.Len(t, sqsMessages.Messages, 1, "Incorrect number of SNS messages sent")
+	assert.Contains(t, *sqsMessages.Messages[0].Body, "balance_enforcement", "Balance enforcement alert was not sent")
+
 	require.Equal(t, 1, len(posts), "No balance enforcement alert was sent")
 	assert.Contains(t, posts[0].Text, "balance_enforcement", "Balance enforcement alert was not sent")
 	assert.Contains(t, posts[0].Text, alertMsg)
@@ -170,4 +189,5 @@ func TestCoolDown(t *testing.T) {
 	time.Sleep(1 * time.Second)
 	posts = ts.TestSlackSvr.SlackAlerts()
 	assert.Equal(t, 1, len(posts), "No alerts should be sent after the transaction is sent")
+
 }

e2e/heuristic_test.go

@@ -31,7 +31,7 @@ import (
 // balance enforcement heuristic session on L2 network.
 func TestBalanceEnforcement(t *testing.T) {
 
-	ts := e2e.CreateSysTestSuite(t)
+	ts := e2e.CreateSysTestSuite(t, "")
 	defer ts.Close()
 
 	alice := ts.Cfg.Secrets.Addresses().Alice
@@ -158,7 +158,7 @@ func TestBalanceEnforcement(t *testing.T) {
 // contract event heuristic session on L1 network.
 func TestContractEvent(t *testing.T) {
 
-	ts := e2e.CreateSysTestSuite(t)
+	ts := e2e.CreateSysTestSuite(t, "")
 	defer ts.Close()
 
 	// The string declaration of the event we want to listen for.
@@ -226,7 +226,7 @@ func TestContractEvent(t *testing.T) {
 // safety heuristic session. This test ensures that an alert is produced in the event
 // of a highly suspicious withdrawal.
 func TestWithdrawalSafetyAllInvariants(t *testing.T) {
-	ts := e2e.CreateSysTestSuite(t)
+	ts := e2e.CreateSysTestSuite(t, "")
 	defer ts.Close()
 
 	opts, err := bind.NewKeyedTransactorWithChainID(ts.Cfg.Secrets.Alice, ts.Cfg.L2ChainIDBig())
@@ -357,7 +357,7 @@ func TestWithdrawalSafetyAllInvariants(t *testing.T) {
 // of a normal tx
 func TestWithdrawalSafetyNoInvariants(t *testing.T) {
 
-	ts := e2e.CreateSysTestSuite(t)
+	ts := e2e.CreateSysTestSuite(t, "")
 	defer ts.Close()
 
 	ids, err := ts.App.BootStrap([]*models.SessionRequestParams{
@@ -439,7 +439,7 @@ func TestWithdrawalSafetyNoInvariants(t *testing.T) {
 // TestFaultDetector ... Ensures that an alert is produced in the presence of a faulty L2Output root
 // on the L1 Optimism portal contract.
 func TestFaultDetector(t *testing.T) {
-	ts := e2e.CreateSysTestSuite(t)
+	ts := e2e.CreateSysTestSuite(t, "")
 	defer ts.Close()
 
 	// Generate transactor opts

e2e/setup.go

@@ -6,6 +6,9 @@ import (
 	"os"
 	"testing"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/sqs"
 	"github.com/base-org/pessimism/internal/alert"
 	"github.com/base-org/pessimism/internal/api/server"
 	"github.com/base-org/pessimism/internal/app"
@@ -19,11 +22,11 @@ import (
 	"github.com/base-org/pessimism/internal/state"
 	"github.com/base-org/pessimism/internal/subsystem"
 	ix_node "github.com/ethereum-optimism/optimism/indexer/node"
-	"github.com/golang/mock/gomock"
-
 	op_e2e "github.com/ethereum-optimism/optimism/op-e2e"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/golang/mock/gomock"
+	"go.uber.org/zap"
 )
 
 // SysTestSuite ... Stores all the information needed to run an e2e system test
@@ -51,7 +54,7 @@ type SysTestSuite struct {
 }
 
 // CreateSysTestSuite ... Creates a new SysTestSuite
-func CreateSysTestSuite(t *testing.T) *SysTestSuite {
+func CreateSysTestSuite(t *testing.T, topicArn string) *SysTestSuite {
 	t.Log("Creating system test suite")
 	ctx := context.Background()
 	logging.New(core.Development)
@@ -112,11 +115,14 @@ func CreateSysTestSuite(t *testing.T) *SysTestSuite {
 
 	pagerdutyServer := NewTestPagerDutyServer("127.0.0.1", 0)
 
+	setAwsVars(t)
+
 	slackURL := fmt.Sprintf("http://127.0.0.1:%d", slackServer.Port)
 	pagerdutyURL := fmt.Sprintf("http://127.0.0.1:%d", pagerdutyServer.Port)
 
 	appCfg.AlertConfig.PagerdutyAlertEventsURL = pagerdutyURL
 	appCfg.AlertConfig.RoutingParams = DefaultRoutingParams(core.StringFromEnv(slackURL))
+	appCfg.AlertConfig.SNSConfig.TopicArn = topicArn
 
 	pess, kill, err := app.NewPessimismApp(ctx, appCfg)
 	if err != nil {
@@ -165,6 +171,9 @@ func DefaultTestConfig() *config.Config {
 		AlertConfig: &alert.Config{
 			PagerdutyAlertEventsURL: "",
 			RoutingCfgPath:          "",
+			SNSConfig: &client.SNSConfig{
+				Endpoint: "http://localhost:4566",
+			},
 		},
 		EngineConfig: &engine.Config{
 			WorkerCount: workerCount,
@@ -186,6 +195,52 @@ func DefaultTestConfig() *config.Config {
 	}
 }
 
+func setAwsVars(t *testing.T) {
+	awsEnvVariables := map[string]string{
+		"AWS_REGION":            "us-east-1",
+		"AWS_SECRET_ACCESS_KEY": "test",
+		"AWS_ACCESS_KEY_ID":     "test",
+	}
+	for key, value := range awsEnvVariables {
+		if err := os.Setenv(key, value); err != nil {
+			t.Fatalf("Error setting %s environment variable: %s", key, err)
+		}
+	}
+}
+
+func GetSNSMessages(endpoint string, queueName string) (*sqs.ReceiveMessageOutput, error) {
+	sess, err := session.NewSession(&aws.Config{
+		Endpoint: aws.String(endpoint),
+	})
+	if err != nil {
+		logging.NoContext().Error("failed to create AWS session", zap.Error(err))
+		return nil, err
+	}
+
+	svc := sqs.New(sess)
+	urlResult, err := svc.GetQueueUrl(&sqs.GetQueueUrlInput{
+		QueueName: aws.String(queueName),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	queueURL := urlResult.QueueUrl
+	msgResult, err := svc.ReceiveMessage(&sqs.ReceiveMessageInput{
+		QueueUrl:            queueURL,
+		MaxNumberOfMessages: aws.Int64(10),
+		WaitTimeSeconds:     aws.Int64(5),
+		MessageAttributeNames: []*string{
+			aws.String(sqs.QueueAttributeNameAll),
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return msgResult, nil
+}
+
 // DefaultRoutingParams ... Returns a default routing params configuration for testing
 func DefaultRoutingParams(slackURL core.StringFromEnv) *core.AlertRoutingParams {
 	return &core.AlertRoutingParams{

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/Microsoft/go-winio v0.6.1 // indirect
 	github.com/VictoriaMetrics/fastcache v1.10.0 // indirect
 	github.com/ajg/form v1.5.1 // indirect
+	github.com/aws/aws-sdk-go v1.50.3 // indirect
 	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.7.0 // indirect
@@ -50,13 +51,17 @@ require (
 	github.com/decred/dcrd/crypto/blake256 v1.0.1 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect
 	github.com/deepmap/oapi-codegen v1.8.2 // indirect
+	github.com/distribution/reference v0.5.0 // indirect
 	github.com/dlclark/regexp2 v1.7.0 // indirect
+	github.com/docker/docker v25.0.2+incompatible // indirect
+	github.com/docker/go-connections v0.5.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/dop251/goja v0.0.0-20230806174421-c933cf95e127 // indirect
 	github.com/elastic/gosigar v0.14.2 // indirect
 	github.com/ethereum-optimism/go-ethereum-hdwallet v0.1.3 // indirect
 	github.com/ethereum-optimism/superchain-registry/superchain v0.0.0-20231001123245-7b48d3818686 // indirect
 	github.com/ethereum/c-kzg-4844 v0.3.1 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fjl/memsize v0.0.1 // indirect
 	github.com/flynn/noise v1.0.0 // indirect
 	github.com/francoispqt/gojay v1.2.13 // indirect
@@ -65,6 +70,8 @@ require (
 	github.com/getsentry/sentry-go v0.18.0 // indirect
 	github.com/go-chi/chi/v5 v5.0.10 // indirect
 	github.com/go-chi/docgen v1.2.0 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.3+incompatible // indirect
 	github.com/go-stack/stack v1.8.1 // indirect
@@ -74,7 +81,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt/v4 v4.4.2 // indirect
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
-	github.com/google/go-cmp v0.5.9 // indirect
+	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gopacket v1.1.19 // indirect
 	github.com/google/pprof v0.0.0-20230817174616-7a8ec2ada47b // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
@@ -104,6 +111,7 @@ require (
 	github.com/jbenet/goprocess v0.1.4 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/klauspost/compress v1.16.7 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.5 // indirect
 	github.com/koron/go-ssdp v0.0.4 // indirect
@@ -148,6 +156,8 @@ require (
 	github.com/multiformats/go-varint v0.0.7 // indirect
 	github.com/onsi/ginkgo/v2 v2.12.0 // indirect
 	github.com/onsi/gomega v1.28.0 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.0.2 // indirect
 	github.com/opencontainers/runtime-spec v1.1.0 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
@@ -178,6 +188,10 @@ require (
 	github.com/urfave/cli/v2 v2.25.7 // indirect
 	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.47.0 // indirect
+	go.opentelemetry.io/otel v1.22.0 // indirect
+	go.opentelemetry.io/otel/metric v1.22.0 // indirect
+	go.opentelemetry.io/otel/trace v1.22.0 // indirect
 	go.uber.org/dig v1.17.0 // indirect
 	go.uber.org/fx v1.20.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect