Monitoring improvements

cmd/monitoring/README.md

@@ -24,9 +24,7 @@ An example of a compatible json encoded feeds configuration is:
     "contract_address_base58": "2jVYiZgQ5disuAUMxrF1LkUyhZuqvRCrx1LfB555XUUv",
     "transmissions_account_base58": "2jVYiZgQ5disuAUMxrF1LkUyhZuqvRCrx1LfB555XUUv",
     "state_account_base58": "2jVYiZgQ5disuAUMxrF1LkUyhZuqvRCrx1LfB555XUUv",
-
-    "poll_interval_milliseconds": 1000
-  },
+  }
   {
     "name": "link/usd",
     "path": "link-usd",
@@ -38,34 +36,35 @@ An example of a compatible json encoded feeds configuration is:
     "contract_address_base58": "GUnMZPbhxkimy9ssXyPG8rVTPBPFzL24W4vFuxyEZm66",
     "transmissions_account_base58": "GUnMZPbhxkimy9ssXyPG8rVTPBPFzL24W4vFuxyEZm66",
     "state_account_base58": "GUnMZPbhxkimy9ssXyPG8rVTPBPFzL24W4vFuxyEZm66",
-
-    "poll_interval_milliseconds": 1000
   }
 ]
 ```
 
 To build and execute the monitor locally, run:
 
 ```bash
-go run ./cmd/monitoring/*.go \
--solana.rpc_endpoint="http://127.0.0.1:8899" \
--solana.network_name="solana-devnet" \
--solana.network_id="solana-devnet" \
--solana.chain_id="1" \
--kafka.config_set_topic="solana-devnet" \
--kafka.config_set_simplified_topic="solana-devnet" \
--kafka.transmission_topic="solana-devnet" \
--kafka.brokers="localhost:29092" \
--kafka.client_id="solana" \
--kafka.security_protocol="PLAINTEXT" \
--kafka.sasl_mechanism="PLAIN" \
--kafka.sasl_username="" \
--kafka.sasl_password="" \
--schema_registry.url="http://localhost:8989" \
--schema_registry.username="" \
--schema_registry.password="" \
--feeds.file_path="/tmp/feeds.json" \
--http.address="localhost:3000"
+SOLANA_RPC_ENDPOINT="http://127.0.0.1:8899" \
+SOLANA_NETWORK_NAME="solana-devnet" \
+SOLANA_NETWORK_ID="solana-devnet" \
+SOLANA_CHAIN_ID="1" \
+SOLANA_READ_TIMEOUT="2s" \
+SOLANA_POLL_INTERVAL="5s" \
+KAFKA_BROKERS="localhost:29092" \
+KAFKA_CLIENT_ID="solana" \
+KAFKA_SECURITY_PROTOCOL="PLAINTEXT" \
+KAFKA_SASL_MECHANISM="PLAIN" \
+KAFKA_SASL_USERNAME="" \
+KAFKA_SASL_PASSWORD="" \
+KAFKA_CONFIG_SET_TOPIC="config_set" \
+KAFKA_CONFIG_SET_SIMPLIFIED_TOPIC="config_set_simplified" \
+KAFKA_TRANSMISSION_TOPIC="transmission_topic" \
+SCHEMA_REGISTRY_URL="http://localhost:8989" \
+SCHEMA_REGISTRY_USERNAME="" \
+SCHEMA_REGISTRY_PASSWORD="" \
+FEEDS_FILE_PATH="/tmp/feeds.json" \
+HTTP_ADDRESS="localhost:3000" \
+FEATURE_TEST_MODE=true \
+go run ./cmd/monitoring/main.go
 ```
 
 See `go run ./cmd/monitoring/*.go -help` for details.

cmd/monitoring/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/gagliardetto/solana-go/rpc"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/config"
 	"github.com/smartcontractkit/chainlink/core/logger"
 	"go.uber.org/zap/zapcore"
 )
@@ -24,7 +25,7 @@ func main() {
 
 	log := logger.NewLogger(loggerConfig{})
 
-	cfg, err := monitoring.ParseConfig(bgCtx)
+	cfg, err := config.Parse()
 	if err != nil {
 		log.Fatalw("failed to parse configuration", "error", err)
 	}
@@ -51,41 +52,49 @@ func main() {
 	client := rpc.New(cfg.Solana.RPCEndpoint)
 
 	schemaRegistry := monitoring.NewSchemaRegistry(cfg.SchemaRegistry, log)
-	trSchema, err := schemaRegistry.EnsureSchema("transmission-value", monitoring.TransmissionAvroSchema)
-
+	transmissionSchema, err := schemaRegistry.EnsureSchema(cfg.Kafka.TransmissionTopic+"-value", monitoring.TransmissionAvroSchema)
 	if err != nil {
 		log.Fatalw("failed to prepare transmission schema", "error", err)
 	}
-	stSchema, err := schemaRegistry.EnsureSchema(cfg.Kafka.ConfigSetTopic+"-value", monitoring.ConfigSetAvroSchema)
+	configSetSchema, err := schemaRegistry.EnsureSchema(cfg.Kafka.ConfigSetTopic+"-value", monitoring.ConfigSetAvroSchema)
 	if err != nil {
 		log.Fatalf("failed to prepare config_set schema", "error", err)
 	}
-
-	csSimplifiedSchema, err := schemaRegistry.EnsureSchema(cfg.Kafka.ConfigSetSimplifiedTopic+"-value", monitoring.ConfigSetSimplifiedAvroSchema)
+	configSetSimplifiedSchema, err := schemaRegistry.EnsureSchema(cfg.Kafka.ConfigSetSimplifiedTopic+"-value", monitoring.ConfigSetSimplifiedAvroSchema)
 	if err != nil {
 		log.Fatalf("failed to prepare config_set_simplified schema", "error", err)
 	}
+
 	producer, err := monitoring.NewProducer(bgCtx, log.With("component", "producer"), cfg.Kafka)
 	if err != nil {
 		log.Fatalf("failed to create kafka producer", "error", err)
 	}
 
-	var trReader, stReader monitoring.AccountReader
-	if testMode, envVarPresent := os.LookupEnv("TEST_MODE"); envVarPresent && testMode == "enabled" {
-		trReader = monitoring.NewRandomDataReader(bgCtx, wg, "transmission", log.With("component", "rand-reader", "account", "transmissions"))
-		stReader = monitoring.NewRandomDataReader(bgCtx, wg, "state", log.With("component", "rand-reader", "account", "state"))
+	var transmissionReader, stateReader monitoring.AccountReader
+	if cfg.Feature.TestMode {
+		transmissionReader = monitoring.NewRandomDataReader(bgCtx, wg, "transmission", log.With("component", "rand-reader", "account", "transmissions"))
+		stateReader = monitoring.NewRandomDataReader(bgCtx, wg, "state", log.With("component", "rand-reader", "account", "state"))
 	} else {
-		trReader = monitoring.NewTransmissionReader(client)
-		stReader = monitoring.NewStateReader(client)
+		transmissionReader = monitoring.NewTransmissionReader(client)
+		stateReader = monitoring.NewStateReader(client)
 	}
 
 	monitor := monitoring.NewMultiFeedMonitor(
+		cfg.Solana,
+		cfg.Feeds.Feeds,
+
 		log,
-		cfg,
-		trReader, stReader,
-		trSchema, stSchema, csSimplifiedSchema,
+		transmissionReader, stateReader,
 		producer,
 		monitoring.DefaultMetrics,
+
+		cfg.Kafka.ConfigSetTopic,
+		cfg.Kafka.ConfigSetSimplifiedTopic,
+		cfg.Kafka.TransmissionTopic,
+
+		configSetSchema,
+		configSetSimplifiedSchema,
+		transmissionSchema,
 	)
 	wg.Add(1)
 	go func() {

go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/gagliardetto/treeout v0.1.4
 	github.com/golang/protobuf v1.5.2
 	github.com/linkedin/goavro v2.1.0+incompatible
+	github.com/mr-tron/base58 v1.2.0
 	github.com/onsi/ginkgo/v2 v2.0.0-rc2
 	github.com/onsi/gomega v1.17.0
 	github.com/pkg/errors v0.9.1
@@ -145,7 +146,6 @@ require (
 	github.com/monochromegane/go-gitignore v0.0.0-20200626010858-205db1a8cc00 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
 	github.com/mostynb/zstdpool-freelist v0.0.0-20201229113212-927304c0c3b1 // indirect
-	github.com/mr-tron/base58 v1.2.0 // indirect
 	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.0.2 // indirect

pkg/monitoring/account_reader.go

@@ -3,15 +3,12 @@ package monitoring
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"github.com/gagliardetto/solana-go"
 	"github.com/gagliardetto/solana-go/rpc"
 	pkgSolana "github.com/smartcontractkit/chainlink-solana/pkg/solana"
 )
 
-const solanaAccountReadTimeout = 2 * time.Second
-
 // AccountReader is a wrapper on top of *rpc.Client
 type AccountReader interface {
 	Read(ctx context.Context, account solana.PublicKey) (interface{}, error)
@@ -31,8 +28,6 @@ type trReader struct {
 }
 
 func (t *trReader) Read(ctx context.Context, transmissionsAccount solana.PublicKey) (interface{}, error) {
-	ctx, cancel := context.WithTimeout(ctx, solanaAccountReadTimeout)
-	defer cancel()
 	answer, blockNum, err := pkgSolana.GetLatestTransmission(ctx, t.client, transmissionsAccount)
 	return TransmissionEnvelope{answer, blockNum}, err
 }
@@ -51,9 +46,6 @@ type StateEnvelope struct {
 }
 
 func (s *stReader) Read(ctx context.Context, stateAccount solana.PublicKey) (interface{}, error) {
-	ctx, cancel := context.WithTimeout(ctx, solanaAccountReadTimeout)
-	defer cancel()
-
 	state, blockNum, err := pkgSolana.GetState(ctx, s.client, stateAccount)
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch state : %w", err)