From 30170b941367d09e0f463c43c17f78ed1513d54b Mon Sep 17 00:00:00 2001 From: 9seconds Date: Fri, 11 Mar 2022 16:20:11 +0300 Subject: [PATCH] Add iplist_size metric --- README.md | 2 ++ events/event_stream.go | 2 ++ events/event_stream_test.go | 21 +++++++++++++++++++++ events/init.go | 3 +++ events/init_test.go | 4 ++++ events/multi_observer.go | 15 +++++++++++++++ events/noop.go | 1 + events/noop_test.go | 3 +++ example.config.toml | 2 +- internal/cli/run_proxy.go | 35 ++++++++++++++++++++++++++--------- ipblocklist/firehol.go | 36 ++++++++++++++++++++++++------------ ipblocklist/firehol_test.go | 10 ++++++---- mtglib/events.go | 20 ++++++++++++++++++++ mtglib/events_test.go | 9 +++++++++ stats/init.go | 16 ++++++++++++++++ stats/prometheus.go | 16 ++++++++++++++++ stats/prometheus_test.go | 12 ++++++++++++ stats/statsd.go | 9 +++++++++ stats/statsd_test.go | 16 ++++++++++++++++ 19 files changed, 206 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index c1795a7..0740b75 100644 --- a/README.md +++ b/README.md @@ -383,6 +383,7 @@ Here goes a list of metrics with their types but without a prefix. | client_connections | gauge | `ip_family` | Count of processing client connections. | | telegram_connections | gauge | `telegram_ip`, `dc` | Count of connections to Telegram servers. | | domain_fronting_connections | gauge | `ip_family` | Count of connections to fronting domain. | +| iplist_size | gauge | `ip_list` | A size of either allowlist or blocklist in use. | | telegram_traffic | counter | `telegram_ip`, `dc`, `direction` | Count of bytes, transmitted to/from Telegram. | | domain_fronting_traffic | counter | `direction` | Count of bytes, transmitted to/from fronting domain. | | domain_fronting | counter | – | Count of domain fronting events. | @@ -398,3 +399,4 @@ Tag meaning: | dc | | A number of the Telegram DC for a connection. | | telegram_ip | | IP address of the Telegram server. | | direction | `to_client`, `from_client` | A direction of the traffic flow. | +| ip_list | `allowlist`, `blocklist` | A type of the IP list. | diff --git a/events/event_stream.go b/events/event_stream.go index cc971a6..c847bf2 100644 --- a/events/event_stream.go +++ b/events/event_stream.go @@ -102,6 +102,8 @@ func eventStreamProcessor(ctx context.Context, eventChan <-chan mtglib.Event, ob observer.EventConcurrencyLimited(typedEvt) case mtglib.EventReplayAttack: observer.EventReplayAttack(typedEvt) + case mtglib.EventIPListSize: + observer.EventIPListSize(typedEvt) } } } diff --git a/events/event_stream_test.go b/events/event_stream_test.go index 6989c77..da55f24 100644 --- a/events/event_stream_test.go +++ b/events/event_stream_test.go @@ -204,6 +204,27 @@ func (suite *EventStreamTestSuite) TestEventReplayAttack() { time.Sleep(100 * time.Millisecond) } +func (suite *EventStreamTestSuite) TestEventIPListSize() { + evt := mtglib.NewEventIPListSize(10, true) + + for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { + v. + On("EventIPListSize", mock.Anything). + Once(). + Run(func(args mock.Arguments) { + caught, ok := args.Get(0).(mtglib.EventIPListSize) + + suite.True(ok) + suite.Equal(evt.Timestamp(), caught.Timestamp()) + suite.Equal(evt.Size, caught.Size) + suite.Equal(evt.IsBlockList, caught.IsBlockList) + }) + } + + suite.stream.Send(suite.ctx, evt) + time.Sleep(100 * time.Millisecond) +} + func (suite *EventStreamTestSuite) TearDownTest() { suite.stream.Shutdown() suite.ctxCancel() diff --git a/events/init.go b/events/init.go index f754ede..94804dc 100644 --- a/events/init.go +++ b/events/init.go @@ -53,6 +53,9 @@ type Observer interface { // EventReplayAttack reacts on incoming mtglib.EventReplayAttack event. EventReplayAttack(mtglib.EventReplayAttack) + // EventIPListSize reacts on incoming mtglib.EventIPListSize + EventIPListSize(mtglib.EventIPListSize) + // Shutdown stop observer. Default event stream guarantees: // 1. If shutdown is executed, it is executed only once // 2. Observer won't receieve any new message after this diff --git a/events/init_test.go b/events/init_test.go index 1b2e0cd..3fc4ff0 100644 --- a/events/init_test.go +++ b/events/init_test.go @@ -41,6 +41,10 @@ func (o *ObserverMock) EventReplayAttack(evt mtglib.EventReplayAttack) { o.Called(evt) } +func (o *ObserverMock) EventIPListSize(evt mtglib.EventIPListSize) { + o.Called(evt) +} + func (o *ObserverMock) Shutdown() { o.Called() } diff --git a/events/multi_observer.go b/events/multi_observer.go index eeeacae..2b529cd 100644 --- a/events/multi_observer.go +++ b/events/multi_observer.go @@ -130,6 +130,21 @@ func (m multiObserver) EventReplayAttack(evt mtglib.EventReplayAttack) { wg.Wait() } +func (m multiObserver) EventIPListSize(evt mtglib.EventIPListSize) { + wg := &sync.WaitGroup{} + wg.Add(len(m.observers)) + + for _, v := range m.observers { + go func(obs Observer) { + defer wg.Done() + + obs.EventIPListSize(evt) + }(v) + } + + wg.Wait() +} + func (m multiObserver) Shutdown() { for _, v := range m.observers { v.Shutdown() diff --git a/events/noop.go b/events/noop.go index 0582d9f..68ee073 100644 --- a/events/noop.go +++ b/events/noop.go @@ -25,6 +25,7 @@ func (n noopObserver) EventFinish(_ mtglib.EventFinish) func (n noopObserver) EventConcurrencyLimited(_ mtglib.EventConcurrencyLimited) {} func (n noopObserver) EventIPBlocklisted(_ mtglib.EventIPBlocklisted) {} func (n noopObserver) EventReplayAttack(_ mtglib.EventReplayAttack) {} +func (n noopObserver) EventIPListSize(_ mtglib.EventIPListSize) {} func (n noopObserver) Shutdown() {} // NewNoopObserver creates an observer which discards each message. diff --git a/events/noop_test.go b/events/noop_test.go index 369a061..ebd04a9 100644 --- a/events/noop_test.go +++ b/events/noop_test.go @@ -27,6 +27,7 @@ func (suite *NoopTestSuite) SetupSuite() { "concurrency-limited": mtglib.NewEventConcurrencyLimited(), "ip-blacklisted": mtglib.NewEventIPBlocklisted(net.ParseIP("10.0.0.10")), "replay-attack": mtglib.NewEventReplayAttack("connID"), + "ip-list-size": mtglib.NewEventIPListSize(10, true), } suite.ctx = context.Background() } @@ -65,6 +66,8 @@ func (suite *NoopTestSuite) TestObserver() { observer.EventIPBlocklisted(typedEvt) case mtglib.EventReplayAttack: observer.EventReplayAttack(typedEvt) + case mtglib.EventIPListSize: + observer.EventIPListSize(typedEvt) } }) } diff --git a/example.config.toml b/example.config.toml index b4862f6..bdec7b7 100644 --- a/example.config.toml +++ b/example.config.toml @@ -170,7 +170,7 @@ download-concurrency = 2 # You can provider links here (starts with https:// or http://) or # path to a local file, but in this case it should be absolute. urls = [ - # "https://iplists.firehol.org/files/firehol_level1.netset", + "https://iplists.firehol.org/files/firehol_level1.netset", # "/local.file" ] # How often do we need to update a blocklist set. diff --git a/internal/cli/run_proxy.go b/internal/cli/run_proxy.go index eec0fae..ae06a80 100644 --- a/internal/cli/run_proxy.go +++ b/internal/cli/run_proxy.go @@ -1,6 +1,7 @@ package cli import ( + "context" "fmt" "net" "net/url" @@ -85,7 +86,10 @@ func makeAntiReplayCache(conf *config.Config) mtglib.AntiReplayCache { ) } -func makeIPBlocklist(conf config.ListConfig, logger mtglib.Logger, ntw mtglib.Network) (mtglib.IPBlocklist, error) { +func makeIPBlocklist(conf config.ListConfig, + logger mtglib.Logger, + ntw mtglib.Network, + updateCallback ipblocklist.FireholUpdateCallback) (mtglib.IPBlocklist, error) { if !conf.Enabled.Get(false) { return ipblocklist.NewNoop(), nil } @@ -105,7 +109,8 @@ func makeIPBlocklist(conf config.ListConfig, logger mtglib.Logger, ntw mtglib.Ne ntw, conf.DownloadConcurrency.Get(1), remoteURLs, - localFiles) + localFiles, + updateCallback) if err != nil { return nil, fmt.Errorf("incorrect parameters for firehol: %w", err) } @@ -159,12 +164,23 @@ func runProxy(conf *config.Config, version string) error { // nolint: funlen logger.BindJSON("configuration", conf.String()).Debug("configuration") + eventStream, err := makeEventStream(conf, logger) + if err != nil { + return fmt.Errorf("cannot build event stream: %w", err) + } + ntw, err := makeNetwork(conf, version) if err != nil { return fmt.Errorf("cannot build network: %w", err) } - blocklist, err := makeIPBlocklist(conf.Defense.Blocklist, logger.Named("blocklist"), ntw) + blocklist, err := makeIPBlocklist( + conf.Defense.Blocklist, + logger.Named("blocklist"), + ntw, + func(ctx context.Context, size int) { + eventStream.Send(ctx, mtglib.NewEventIPListSize(size, true)) + }) if err != nil { return fmt.Errorf("cannot build ip blocklist: %w", err) } @@ -172,7 +188,13 @@ func runProxy(conf *config.Config, version string) error { // nolint: funlen var whitelist mtglib.IPBlocklist if conf.Defense.Allowlist.Enabled.Get(false) { - whlist, err := makeIPBlocklist(conf.Defense.Allowlist, logger.Named("allowlist"), ntw) + whlist, err := makeIPBlocklist( + conf.Defense.Allowlist, + logger.Named("allowlist"), + ntw, + func(ctx context.Context, size int) { + eventStream.Send(ctx, mtglib.NewEventIPListSize(size, false)) + }) if err != nil { return fmt.Errorf("cannot build ip allowlist: %w", err) } @@ -180,11 +202,6 @@ func runProxy(conf *config.Config, version string) error { // nolint: funlen whitelist = whlist } - eventStream, err := makeEventStream(conf, logger) - if err != nil { - return fmt.Errorf("cannot build event stream: %w", err) - } - opts := mtglib.ProxyOpts{ Logger: logger, Network: ntw, diff --git a/ipblocklist/firehol.go b/ipblocklist/firehol.go index b2c3cce..d6fa3e8 100644 --- a/ipblocklist/firehol.go +++ b/ipblocklist/firehol.go @@ -23,6 +23,10 @@ var ( fireholIPv6DefaultCIDR = net.CIDRMask(128, 128) // nolint: gomnd ) +// FireholUpdateCallback defines a signature of the callback that has to be +// execute when ip list is updated. +type FireholUpdateCallback func(context.Context, int) + // Firehol is IPBlocklist which uses lists from FireHOL: // https://iplists.firehol.org/ // @@ -42,7 +46,8 @@ type Firehol struct { logger mtglib.Logger updateMutex sync.RWMutex - ranger cidranger.Ranger + updateCallback FireholUpdateCallback + ranger cidranger.Ranger blocklists []files.File @@ -110,7 +115,7 @@ func (f *Firehol) update() { wg := &sync.WaitGroup{} wg.Add(len(f.blocklists)) - treeMutex := &sync.Mutex{} + mutex := &sync.Mutex{} ranger := cidranger.NewPCTrieRanger() for _, v := range f.blocklists { @@ -128,7 +133,7 @@ func (f *Firehol) update() { defer fileContent.Close() - if err := f.updateFromFile(treeMutex, ranger, bufio.NewScanner(fileContent)); err != nil { + if err := f.updateFromFile(mutex, ranger, bufio.NewScanner(fileContent)); err != nil { logger.WarningError("update has failed", err) } }(v) @@ -141,6 +146,10 @@ func (f *Firehol) update() { f.ranger = ranger + if f.updateCallback != nil { + f.updateCallback(ctx, ranger.Len()) + } + f.logger.Info("ip list was updated") } @@ -206,7 +215,8 @@ func (f *Firehol) updateParseLine(text string) (*net.IPNet, error) { func NewFirehol(logger mtglib.Logger, network mtglib.Network, downloadConcurrency uint, urls []string, - localFiles []string) (*Firehol, error) { + localFiles []string, + updateCallback FireholUpdateCallback) (*Firehol, error) { blocklists := []files.File{} for _, v := range localFiles { @@ -229,12 +239,13 @@ func NewFirehol(logger mtglib.Logger, network mtglib.Network, blocklists = append(blocklists, file) } - return NewFireholFromFiles(logger, downloadConcurrency, blocklists) + return NewFireholFromFiles(logger, downloadConcurrency, blocklists, updateCallback) } func NewFireholFromFiles(logger mtglib.Logger, downloadConcurrency uint, - blocklists []files.File) (*Firehol, error) { + blocklists []files.File, + updateCallback FireholUpdateCallback) (*Firehol, error) { if downloadConcurrency == 0 { downloadConcurrency = DefaultFireholDownloadConcurrency } @@ -243,11 +254,12 @@ func NewFireholFromFiles(logger mtglib.Logger, ctx, cancel := context.WithCancel(context.Background()) return &Firehol{ - ctx: ctx, - ctxCancel: cancel, - logger: logger.Named("firehol"), - ranger: cidranger.NewPCTrieRanger(), - workerPool: workerPool, - blocklists: blocklists, + ctx: ctx, + ctxCancel: cancel, + logger: logger.Named("firehol"), + ranger: cidranger.NewPCTrieRanger(), + workerPool: workerPool, + blocklists: blocklists, + updateCallback: updateCallback, }, nil } diff --git a/ipblocklist/firehol_test.go b/ipblocklist/firehol_test.go index 9d26c6d..416d43e 100644 --- a/ipblocklist/firehol_test.go +++ b/ipblocklist/firehol_test.go @@ -67,7 +67,8 @@ func (suite *FireholTestSuite) TearDownSuite() { func (suite *FireholTestSuite) TestLocalFail() { blocklist, err := ipblocklist.NewFirehol(logger.NewNoopLogger(), suite.networkMock, 2, - nil, []string{filepath.Join("testdata", "broken_ipset.ipset")}) + nil, []string{filepath.Join("testdata", "broken_ipset.ipset")}, + nil) suite.NoError(err) @@ -85,7 +86,8 @@ func (suite *FireholTestSuite) TestLocalFail() { func (suite *FireholTestSuite) TestLocalOk() { blocklist, err := ipblocklist.NewFirehol(logger.NewNoopLogger(), suite.networkMock, 2, - nil, []string{filepath.Join("testdata", "good_ipset.ipset")}) + nil, []string{filepath.Join("testdata", "good_ipset.ipset")}, + nil) suite.NoError(err) @@ -103,7 +105,7 @@ func (suite *FireholTestSuite) TestLocalOk() { func (suite *FireholTestSuite) TestRemoteFail() { blocklist, err := ipblocklist.NewFirehol(logger.NewNoopLogger(), suite.networkMock, 2, - []string{"https://google.com"}, nil) + []string{"https://google.com"}, nil, nil) suite.NoError(err) @@ -127,7 +129,7 @@ func (suite *FireholTestSuite) TestMixed() { suite.httpServer.URL, }, []string{ filepath.Join("testdata", "good_ipset.ipset"), - }) + }, nil) suite.NoError(err) diff --git a/mtglib/events.go b/mtglib/events.go index 1bc6fb7..fdc387b 100644 --- a/mtglib/events.go +++ b/mtglib/events.go @@ -92,6 +92,15 @@ type EventReplayAttack struct { eventBase } +// EventIPListSize is emitted when mtg updates a contents of the ip lists: +// allowlist or blocklist. +type EventIPListSize struct { + eventBase + + Size int + IsBlockList bool +} + // NewEventStart creates a new EventStart event. func NewEventStart(streamID string, remoteIP net.IP) EventStart { return EventStart{ @@ -176,3 +185,14 @@ func NewEventReplayAttack(streamID string) EventReplayAttack { }, } } + +// NewEventIPListSize creates a new EventIPListSize event. +func NewEventIPListSize(size int, isBlockList bool) EventIPListSize { + return EventIPListSize{ + eventBase: eventBase{ + timestamp: time.Now(), + }, + Size: size, + IsBlockList: isBlockList, + } +} diff --git a/mtglib/events_test.go b/mtglib/events_test.go index bb88b0a..56da979 100644 --- a/mtglib/events_test.go +++ b/mtglib/events_test.go @@ -69,6 +69,15 @@ func (suite *EventsTestSuite) TestEventReplayAttack() { suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) } +func (suite *EventsTestSuite) TestEventIPListSize() { + evt := mtglib.NewEventIPListSize(10, false) + + suite.Empty(evt.StreamID()) + suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) + suite.Equal(10, evt.Size) + suite.False(evt.IsBlockList) +} + func TestEvents(t *testing.T) { t.Parallel() suite.Run(t, &EventsTestSuite{}) diff --git a/stats/init.go b/stats/init.go index 2154c60..1bf33df 100644 --- a/stats/init.go +++ b/stats/init.go @@ -89,6 +89,13 @@ const ( // Type: counter MetricReplayAttacks = "replay_attacks" + // MetricIPListSize defines a metric for the size of the the ip list. + // + // Type: gauge + // Tags: + // ip_list | 'allowlist' or 'blocklist' + MetricIPListSize = "iplist_size" + // TagIPFamily defines a name of the 'ip_family' tag and all values. TagIPFamily = "ip_family" @@ -114,4 +121,13 @@ const ( // TagDirectionFromClient defines that traffic is sent from a client to // Telegram. TagDirectionFromClient = "from_client" + + // TagIPList defines a name of the 'ip_list' and all values. + TagIPList = "ip_list" + + // TagIPListAllow defines a value of 'ip_list' of allowlist. + TagIPListAllow = "allowlist" + + // TagIPListBlock defines a value of 'ip_list' of blocklist. + TagIPListBlock = "blocklist" ) diff --git a/stats/prometheus.go b/stats/prometheus.go index 97c6e97..17882df 100644 --- a/stats/prometheus.go +++ b/stats/prometheus.go @@ -118,6 +118,15 @@ func (p prometheusProcessor) EventReplayAttack(_ mtglib.EventReplayAttack) { p.factory.metricReplayAttacks.Inc() } +func (p prometheusProcessor) EventIPListSize(evt mtglib.EventIPListSize) { + tag := TagIPListBlock + if !evt.IsBlockList { + tag = TagIPListAllow + } + + p.factory.metricIPListSize.WithLabelValues(tag).Set(float64(evt.Size)) +} + func (p prometheusProcessor) Shutdown() { for k, v := range p.streams { releaseStreamInfo(v) @@ -137,6 +146,7 @@ type PrometheusFactory struct { metricClientConnections *prometheus.GaugeVec metricTelegramConnections *prometheus.GaugeVec metricDomainFrontingConnections *prometheus.GaugeVec + metricIPListSize *prometheus.GaugeVec metricTelegramTraffic *prometheus.CounterVec metricDomainFrontingTraffic *prometheus.CounterVec @@ -197,6 +207,11 @@ func NewPrometheus(metricPrefix, httpPath string) *PrometheusFactory { // nolint Name: MetricDomainFrontingConnections, Help: "A number of connections which talk to front domain.", }, []string{TagIPFamily}), + metricIPListSize: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: metricPrefix, + Name: MetricIPListSize, + Help: "A size of the ip list (blocklist or allowlist)", + }, []string{TagIPList}), metricTelegramTraffic: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: metricPrefix, @@ -234,6 +249,7 @@ func NewPrometheus(metricPrefix, httpPath string) *PrometheusFactory { // nolint registry.MustRegister(factory.metricClientConnections) registry.MustRegister(factory.metricTelegramConnections) registry.MustRegister(factory.metricDomainFrontingConnections) + registry.MustRegister(factory.metricIPListSize) registry.MustRegister(factory.metricTelegramTraffic) registry.MustRegister(factory.metricDomainFrontingTraffic) diff --git a/stats/prometheus_test.go b/stats/prometheus_test.go index 6e44c0b..e530095 100644 --- a/stats/prometheus_test.go +++ b/stats/prometheus_test.go @@ -169,6 +169,18 @@ func (suite *PrometheusTestSuite) TestEventReplayAttack() { suite.Contains(data, `mtg_replay_attacks 1`) } +func (suite *PrometheusTestSuite) TestEventIPListSize() { + suite.prometheus.EventIPListSize(mtglib.NewEventIPListSize(10, false)) + suite.prometheus.EventIPListSize(mtglib.NewEventIPListSize(3, true)) + + time.Sleep(100 * time.Millisecond) + + data, err := suite.Get() + suite.NoError(err) + suite.Contains(data, `mtg_iplist_size{ip_list="allowlist"} 10`) + suite.Contains(data, `mtg_iplist_size{ip_list="blocklist"} 3`) +} + func TestPrometheus(t *testing.T) { t.Parallel() suite.Run(t, &PrometheusTestSuite{}) diff --git a/stats/statsd.go b/stats/statsd.go index fcf8191..bd3f6fb 100644 --- a/stats/statsd.go +++ b/stats/statsd.go @@ -121,6 +121,15 @@ func (s statsdProcessor) EventReplayAttack(_ mtglib.EventReplayAttack) { s.client.Incr(MetricReplayAttacks, 1) } +func (s statsdProcessor) EventIPListSize(evt mtglib.EventIPListSize) { + tag := TagIPListBlock + if !evt.IsBlockList { + tag = TagIPListAllow + } + + s.client.Gauge(MetricIPListSize, int64(evt.Size), statsd.StringTag(TagIPList, tag)) +} + func (s statsdProcessor) Shutdown() { events := make([]mtglib.EventFinish, 0, len(s.streams)) diff --git a/stats/statsd_test.go b/stats/statsd_test.go index d810c42..f3eaafd 100644 --- a/stats/statsd_test.go +++ b/stats/statsd_test.go @@ -196,6 +196,22 @@ func (suite *StatsdTestSuite) TestEventReplayAttack() { suite.Equal("mtg.replay_attacks:1|c", suite.statsdServer.String()) } +func (suite *StatsdTestSuite) TestEventIPListSizeAllowlist() { + suite.statsd.EventIPListSize(mtglib.NewEventIPListSize(10, false)) + + time.Sleep(statsdSleepTime) + suite.Contains(suite.statsdServer.String(), "mtg.iplist_size:10|g") + suite.Contains(suite.statsdServer.String(), "allowlist") +} + +func (suite *StatsdTestSuite) TestEventIPListSizeBlocklist() { + suite.statsd.EventIPListSize(mtglib.NewEventIPListSize(10, true)) + + time.Sleep(statsdSleepTime) + suite.Contains(suite.statsdServer.String(), "mtg.iplist_size:10|g") + suite.Contains(suite.statsdServer.String(), "blocklist") +} + func TestStatsd(t *testing.T) { t.Parallel() suite.Run(t, &StatsdTestSuite{})