Publish SNS event on new alert

.github/workflows/hygeine.yml

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

Makefile

@@ -31,7 +31,8 @@ test:
 
 .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
 
 .PHONY: lint
 lint:

config.env.template

@@ -32,6 +32,8 @@ P0_PAGERDUTY_ALERT_EVENTS_URL=
 P1_PAGERDUTY_INTEGRATION_KEY=
 P1_PAGERDUTY_ALERT_EVENTS_URL=
 
+SNS_TOPIC_ARN=
+
 # 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
+    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-setup.sh:/etc/localstack/init/ready.d/script.sh"

e2e/alerting_test.go

@@ -3,6 +3,7 @@ package e2e_test
 import (
 	"context"
 	"math/big"
+
 	"testing"
 	"time"
 
@@ -17,11 +18,17 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+const (
+	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 +80,12 @@ func TestMultiDirectiveRouting(t *testing.T) {
 		return height != nil && height.Uint64() > receipt.BlockNumber.Uint64(), nil
 	}))
 
+	sqsMessages, err := e2e.GetMessages(ts.AppCfg.AlertConfig.SNSConfig.Endpoint, "multi-directive-test-queue")
+	require.NoError(t, err)
+
+	assert.Len(t, sqsMessages.Messages, 1, "Incorrect number of SNS messages sent")
+	assert.Contains(t, *sqsMessages.Messages[0].Body, "contract_event", "System contract event alert was not sent")
+
 	slackPosts := ts.TestSlackSvr.SlackAlerts()
 	pdPosts := ts.TestPagerDutyServer.PagerDutyAlerts()
 
@@ -90,7 +103,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 +162,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 {
@@ -159,15 +172,23 @@ func TestCoolDown(t *testing.T) {
 		return height != nil && height.Uint64() > receipt.BlockNumber.Uint64(), nil
 	}))
 
+	time.Sleep(1 * time.Second)
+
 	// Check that the balance enforcement was triggered using the mocked server cache.
 	posts := ts.TestSlackSvr.SlackAlerts()
 
 	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)
 
+	sqsMessages, err := e2e.GetMessages(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")
+
 	// Ensure that no new alerts are sent for provided cooldown period.
 	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,16 @@ func CreateSysTestSuite(t *testing.T) *SysTestSuite {
 
 	pagerdutyServer := NewTestPagerDutyServer("127.0.0.1", 0)
 
+	os.Setenv("AWS_REGION", "us-east-1")       //nolint:tenv // Cannot use t.SetEnv in parallel tests
+	os.Setenv("AWS_SECRET_ACCESS_KEY", "test") //nolint:tenv // Cannot use t.SetEnv in parallel tests
+	os.Setenv("AWS_ACCESS_KEY_ID", "test")     //nolint:tenv // Cannot use t.SetEnv in parallel tests
+
 	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 +173,9 @@ func DefaultTestConfig() *config.Config {
 		AlertConfig: &alert.Config{
 			PagerdutyAlertEventsURL: "",
 			RoutingCfgPath:          "",
+			SNSConfig: &client.SNSConfig{
+				Endpoint: "http://localhost:4566",
+			},
 		},
 		EngineConfig: &engine.Config{
 			WorkerCount: workerCount,
@@ -186,6 +197,39 @@ func DefaultTestConfig() *config.Config {
 	}
 }
 
+func GetMessages(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