Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add support for JuelsPerFeeCoin, AggregatorRoundID and TxResults #182

Merged
merged 3 commits into from
Feb 7, 2022
Merged

Add support for JuelsPerFeeCoin, AggregatorRoundID and TxResults #182

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
10f2fc9
Use the new Entrypoint abstraction from the core relay monitor
topliceanu Feb 4, 2022
53f167c
add JuelsPerFeeCoin and AggregatorRoundID in Envelope. Introduce a
topliceanu Feb 4, 2022
2d511a4
bump chainlink-relay/pkg/monitoring and fix linter issues
topliceanu Feb 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
139 changes: 32 additions & 107 deletions cmd/monitoring/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,9 @@ package main

import (
"context"
"os"
"os/signal"
"sync"
"syscall"

"github.com/gagliardetto/solana-go/rpc"
relayMonitoring "github.com/smartcontractkit/chainlink-relay/pkg/monitoring"
relayConfig "github.com/smartcontractkit/chainlink-relay/pkg/monitoring/config"
"github.com/smartcontractkit/chainlink-solana/pkg/monitoring"
"github.com/smartcontractkit/chainlink/core/logger"
"go.uber.org/zap/zapcore"
Expand All @@ -23,120 +19,47 @@ func main() {
log.Fatalw("failed to parse solana-specific config", "error", err)
}

sourceFactory := monitoring.NewSolanaSourceFactory(
chainConfig,
logWrapper{coreLog.With("component", "source")},
)

wg := &sync.WaitGroup{}
defer wg.Wait()
client := rpc.New(chainConfig.RPCEndpoint)

bgCtx, cancelBgCtx := context.WithCancel(context.Background())
defer cancelBgCtx()

cfg, err := relayConfig.Parse()
if err != nil {
log.Fatalw("failed to parse generic configuration", "error", err)
}

schemaRegistry := relayMonitoring.NewSchemaRegistry(cfg.SchemaRegistry, log)

transmissionSchema, err := schemaRegistry.EnsureSchema(cfg.Kafka.TransmissionTopic+"-value", relayMonitoring.TransmissionAvroSchema)
if err != nil {
log.Fatalw("failed to prepare transmission schema", "error", err)
}
configSetSimplifiedSchema, err := schemaRegistry.EnsureSchema(cfg.Kafka.ConfigSetSimplifiedTopic+"-value", relayMonitoring.ConfigSetSimplifiedAvroSchema)
if err != nil {
log.Fatalw("failed to prepare config_set_simplified schema", "error", err)
}
envelopeSourceFactory := monitoring.NewEnvelopeSourceFactory(
client,
logWrapper{coreLog.With("component", "source-envelope")},
)
txResultsSourceFactory := monitoring.NewTxResultsSourceFactory(
client,
logWrapper{coreLog.With("component", "source-txresults")},
)

producer, err := relayMonitoring.NewProducer(bgCtx, log.With("component", "producer"), cfg.Kafka)
entrypoint, err := relayMonitoring.NewEntrypoint(
context.Background(),
log,
chainConfig,
envelopeSourceFactory,
txResultsSourceFactory,
monitoring.SolanaFeedParser,
)
if err != nil {
log.Fatalw("failed to create kafka producer", "error", err)
log.Fatalw("failed to build entrypoint", "error", err)
return
}

if cfg.Feature.TestOnlyFakeReaders {
sourceFactory = relayMonitoring.NewRandomDataSourceFactory(bgCtx, wg, log.With("component", "rand-source"))
}

balancesSourceFactory := monitoring.NewBalancesSourceFactory(chainConfig, log.With("component", "balances-source"))
if cfg.Feature.TestOnlyFakeReaders {
balancesSourceFactory := monitoring.NewBalancesSourceFactory(
client,
log.With("component", "source-balances"),
)
if entrypoint.Config.Feature.TestOnlyFakeReaders {
balancesSourceFactory = monitoring.NewFakeBalancesSourceFactory(log.With("component", "fake-balances-source"))
}
entrypoint.SourceFactories = append(entrypoint.SourceFactories, balancesSourceFactory)

metrics := relayMonitoring.DefaultMetrics

prometheusExporterFactory := relayMonitoring.NewPrometheusExporterFactory(
log.With("component", "prometheus-exporter"),
metrics,
)
kafkaExporterFactory := relayMonitoring.NewKafkaExporterFactory(
log.With("component", "kafka-exporter"),
producer,

transmissionSchema,
configSetSimplifiedSchema,

cfg.Kafka.TransmissionTopic,
cfg.Kafka.ConfigSetSimplifiedTopic,
)

balancesPrometheusExporterFactory := monitoring.NewPrometheusExporterFactory(
balancesExporterFactory := monitoring.NewPrometheusExporterFactory(
log.With("component", "balances-prometheus-exporter"),
monitoring.DefaultMetrics,
)
entrypoint.ExporterFactories = append(entrypoint.ExporterFactories, balancesExporterFactory)

monitor := relayMonitoring.NewMultiFeedMonitor(
chainConfig,
log,
[]relayMonitoring.SourceFactory{sourceFactory, balancesSourceFactory},
[]relayMonitoring.ExporterFactory{prometheusExporterFactory, kafkaExporterFactory, balancesPrometheusExporterFactory},
)

rddSource := relayMonitoring.NewRDDSource(cfg.Feeds.URL, monitoring.SolanaFeedParser)
if cfg.Feature.TestOnlyFakeRdd {
// Generate between 2 and 10 random feeds every RDDPollInterval.
rddSource = monitoring.NewFakeRDDSource(2, 10)
}
rddPoller := relayMonitoring.NewSourcePoller(
rddSource,
log.With("component", "rdd-poller"),
cfg.Feeds.RDDPollInterval,
cfg.Feeds.RDDReadTimeout,
0, // no buffering!
)
wg.Add(1)
go func() {
defer wg.Done()
rddPoller.Run(bgCtx)
}()

manager := relayMonitoring.NewManager(
log.With("component", "manager"),
rddPoller,
)
wg.Add(1)
go func() {
defer wg.Done()
manager.Run(bgCtx, monitor.Run)
}()

// Configure HTTP server
http := relayMonitoring.NewHttpServer(bgCtx, cfg.Http.Address, log.With("component", "http-server"))
http.Handle("/metrics", metrics.HTTPHandler())
http.Handle("/debug", manager.HTTPHandler())
wg.Add(1)
go func() {
defer wg.Done()
http.Run(bgCtx)
}()

// Handle signals from the OS
osSignalsCh := make(chan os.Signal, 1)
signal.Notify(osSignalsCh, syscall.SIGINT, syscall.SIGTERM)
sig := <-osSignalsCh
log.Infow("received signal. Stopping", "signal", sig)

entrypoint.Run()
log.Infow("monitor stopped")
}

// logger config
Expand Down Expand Up @@ -165,6 +88,8 @@ func (l loggerConfig) LogUnixTimestamps() bool {
return false // log timestamp in ISO8601
}

// adapt core logger to monitoring logger.

type logWrapper struct {
logger.Logger
}
Expand Down
4 changes: 2 additions & 2 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,10 @@ require (
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.12.0
github.com/rs/zerolog v1.26.1
github.com/smartcontractkit/chainlink-relay v0.0.0-20220128165504-9a2d2530b33e
github.com/smartcontractkit/chainlink-relay v0.0.0-20220204181714-04bbff42571d
github.com/smartcontractkit/integrations-framework v1.0.42
go.uber.org/atomic v1.9.0
go.uber.org/multierr v1.7.0
go.uber.org/zap v1.19.1
golang.org/x/crypto v0.0.0-20211215165025-cf75a172585e
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
Expand Down Expand Up @@ -184,7 +185,6 @@ require (
github.com/xlab/treeprint v1.1.0 // indirect
go.opencensus.io v0.23.0 // indirect
go.starlark.net v0.0.0-20211013185944-b0039bd2cfe3 // indirect
go.uber.org/multierr v1.7.0 // indirect
go.uber.org/ratelimit v0.2.0 // indirect
golang.org/x/net v0.0.0-20211209124913-491a49abca63 // indirect
golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
Expand Down
Loading