Publish SNS event on new alert (#200)

.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