Merge pull request #255 from 9seconds/iplistsize

Add iplist_size metric

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.                        |

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)
 			}
 		}
 	}

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()

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

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()
 }

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()

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.

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)
 			}
 		})
 	}

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.

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,

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
 }

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)
 

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,
+	}
+}

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{})

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"
 )

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)

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{})

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))
 

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{})