Ability to filter reported channels (#19)

* feat: simple filter feature

* style: naming, use fsnotify

* fix: watcher issues

* fix: simplify and remove periodic timer for refreshing file

* feat: additional options

* fix: typo

* docs: update text

README.md

@@ -20,26 +20,25 @@ USAGE:
    balance-agent [global options] command [command options] [arguments...]
 
 VERSION:
-   v0.0.22
+   v0.0.38
 
 COMMANDS:
    help, h  Shows a list of commands or help for one command
 
 GLOBAL OPTIONS:
-   --apikey value             api key
-   --rpcserver value          host:port of ln daemon (default: "localhost:10009")
-   --lnddir value             path to lnd's base directory (default: "/home/user/lnd")
-   --tlscertpath value        path to TLS certificate (default: "/home/user/lnd/tls.cert")
-   --chain value, -c value    the chain lnd is running on e.g. bitcoin (default: "bitcoin")
-   --network value, -n value  the network lnd is running on e.g. mainnet, testnet, etc. (default: "mainnet")
-   --macaroonpath value       path to macaroon file
-   --allowedentropy value     allowed entropy in bits for channel balances (default: 64)
-   --interval value           interval to poll - 10s, 1m, 10m or 1h (default: "10s")
-   --private                  report private data as well (default: false)
-   --preferipv4               If you have the choice between IPv6 and IPv4 prefer IPv4 (default: false)
-   --verbosity value          log level for V logs (default: 0)
-   --help, -h                 show help
-   --version, -v              print the version
+   --allowedentropy value      allowed entropy in bits for channel balances (default: 64)
+   --apikey value              api key
+   --interval value            interval to poll - 10s, 1m, 10m or 1h (default: "10s")
+   --lnddir value              path to lnd's base directory (default: "/home/user/.lnd")
+   --macaroonpath value        path to macaroon file
+   --private                   report private data as well (default: false)
+   --preferipv4                If you have the choice between IPv6 and IPv4 prefer IPv4 (default: false)
+   --rpcserver value           host:port of ln daemon (default: "localhost:10009")
+   --tlscertpath value         path to TLS certificate (default: "/home/user/.lnd/tls.cert")
+   --channel-whitelist value   Path to file containing a whitelist of channels
+   --verbosity value           log level for V logs (default: 0)
+   --help, -h                  show help
+   --version, -v               print the version
 ```
 
 It tries the best to have sane defaults so you can just start it up on your node without further hassle.
@@ -129,11 +128,39 @@ Usage:
 docker run -v /tmp:/tmp -e API_KEY=changeme ghcr.io/bolt-observer/agent:v0.0.35
 ```
 
+## Filtering on agent side
+
+You can limit what channnels are reported using `--channel-whitelist` option. It specifies a local file path to be used as a whitelist of what channels to report.
+When adding `--private` to `--channel-whitelist` this means every private channel AND whatever is listed in the file. There is also `--public` to allow all public channels.
+Using the options without `--channel-whitelist` makes no sense since by default all public channels are reported however it has to be explicit with `--channel-whitelist` in order
+to automatically allow all public channels (beside what is allowed through the file).
+If you set `--private` and `--public` then no matter what you add to the `--channel-whitelist` file everything will be reported.
+
+The file looks like this:
+
+```
+# Comments start with a # character
+# You can list pubkeys for example:
+0288037d3f0bdcfb240402b43b80cdc32e41528b3e2ebe05884aff507d71fca71a # bolt.observer
+# which means any channel where peer pubkey is this
+# or you can specify a specific short channel id e.g.,
+759930760125546497
+# too, invalid lines like
+whatever
+# will be ignored (and logged as a warning, aliases also don't work!)
+# Validity for channel id is not checked (it just has to be numeric), thus:
+1337
+# is perfectly valid (altho it won't match and thus allow the reporting of
+# any additional channel).
+# Empty files means nothing - in whitelist context: do not report anything.
+```
+
 ## Components
 
 Internally we use:
 * [channelchecker](./channelchecker): an abstraction for checking all channels
 * [nodeinfo](./nodeinfo): this can basically report `lncli getnodeinfo` for your node  - it is used by the agent so we have a full view of node info & channels
+* [filter](./filter): this is used to filter specific channels on the agent side
 * [checkermonitoring](./checkermonitoring): is used for reporting metrics via Graphite (not used directly in balance-agent here)
 * [lightning_api](./lightning_api): an abstraction around lightning node API (that furthermore heavily depends on common code from [lnd](https://github.com/lightningnetwork/lnd))
 

channelchecker/channelchecker.go

@@ -14,6 +14,7 @@ import (
 
 	checkermonitoring "github.com/bolt-observer/agent/checkermonitoring"
 	entities "github.com/bolt-observer/agent/entities"
+	"github.com/bolt-observer/agent/filter"
 	api "github.com/bolt-observer/agent/lightning_api"
 	common_entities "github.com/bolt-observer/go_common/entities"
 	utils "github.com/bolt-observer/go_common/utils"
@@ -109,6 +110,12 @@ func (c *ChannelChecker) Subscribe(
 		settings.NoopInterval = c.keepAliveInterval
 	}
 
+	if settings.Filter == nil {
+		glog.V(3).Infof("Filter was nil, allowing everything")
+		f, _ := filter.NewAllowAllFilter()
+		settings.Filter = f
+	}
+
 	c.globalSettings.Set(info.IdentityPubkey+uniqueId, Settings{
 		identifier:     entities.NodeIdentifier{Identifier: pubKey, UniqueId: uniqueId},
 		settings:       settings,
@@ -140,6 +147,11 @@ func (c *ChannelChecker) GetState(
 		return nil, errors.New("invalid pubkey")
 	}
 
+	if settings.Filter == nil {
+		f, _ := filter.NewAllowAllFilter()
+		settings.Filter = f
+	}
+
 	resp, err := c.checkOne(entities.NodeIdentifier{Identifier: pubKey, UniqueId: uniqueId}, getApi, settings, true, false)
 	if err != nil {
 		return nil, err
@@ -156,7 +168,9 @@ func (c *ChannelChecker) getChannelList(
 	api api.LightingApiCalls,
 	info *api.InfoApi,
 	precisionBits int,
-	allowPrivateChans bool) ([]entities.ChannelBalance, SetOfChanIds, error) {
+	allowPrivateChans bool,
+	filter filter.FilterInterface,
+) ([]entities.ChannelBalance, SetOfChanIds, error) {
 
 	defer c.monitoring.MetricsTimer("channellist", map[string]string{"pubkey": info.IdentityPubkey})()
 
@@ -180,6 +194,12 @@ func (c *ChannelChecker) getChannelList(
 
 	for _, channel := range channels.Channels {
 		if channel.Private && !allowPrivateChans {
+			glog.V(3).Infof("Skipping private channel %v", channel.ChanId)
+			continue
+		}
+
+		if !filter.AllowChanId(channel.ChanId) && !filter.AllowPubKey(channel.RemotePubkey) && !filter.AllowSpecial(channel.Private) {
+			glog.V(3).Infof("Filtering channel %v", channel.ChanId)
 			continue
 		}
 
@@ -474,7 +494,7 @@ func (c *ChannelChecker) checkOne(
 		identifier.Identifier = info.IdentityPubkey
 	}
 
-	channelList, set, err := c.getChannelList(api, info, settings.AllowedEntropy, settings.AllowPrivateChannels)
+	channelList, set, err := c.getChannelList(api, info, settings.AllowedEntropy, settings.AllowPrivateChannels, settings.Filter)
 	if err != nil {
 		c.monitoring.MetricsReport("checkone", "failure", map[string]string{"pubkey": pubkey})
 		return nil, err

channelchecker/channelchecker_test.go

@@ -14,6 +14,7 @@ import (
 
 	miniredis "github.com/alicebob/miniredis/v2"
 	agent_entities "github.com/bolt-observer/agent/entities"
+	"github.com/bolt-observer/agent/filter"
 	lightning_api "github.com/bolt-observer/agent/lightning_api"
 	entities "github.com/bolt-observer/go_common/entities"
 	utils "github.com/bolt-observer/go_common/utils"
@@ -216,6 +217,129 @@ func TestBasicFlow(t *testing.T) {
 	}
 }
 
+func TestBasicFlowFilterOne(t *testing.T) {
+	pubKey, api, d := initTest(t)
+
+	d.HttpApi.DoFunc = func(req *http.Request) (*http.Response, error) {
+		contents := ""
+		if strings.Contains(req.URL.Path, "v1/getinfo") {
+			contents = getInfoJson("02b67e55fb850d7f7d77eb71038362bc0ed0abd5b7ee72cc4f90b16786c69b9256")
+		} else if strings.Contains(req.URL.Path, "v1/channels") {
+			contents = getChannelJson(1337, false, true)
+		}
+
+		r := ioutil.NopCloser(bytes.NewReader([]byte(contents)))
+
+		return &http.Response{
+			StatusCode: 200,
+			Body:       r,
+		}, nil
+	}
+
+	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(15*time.Second))
+
+	c := NewDefaultChannelChecker(ctx, time.Duration(0), true, false, nil)
+	// Make everything a bit faster
+	c.OverrideLoopInterval(1 * time.Second)
+	was_called := false
+
+	f, _ := filter.NewUnitTestFilter()
+	fd := f.(*filter.UnitTestFilter)
+	fd.AddAllowChanId(1)
+	fd.AddAllowChanId(1337)
+
+	c.Subscribe(
+		pubKey, "random_id",
+		func() lightning_api.LightingApiCalls { return api },
+		agent_entities.ReportingSettings{
+			AllowedEntropy:       64,
+			PollInterval:         agent_entities.SECOND,
+			AllowPrivateChannels: true,
+			Filter:               f,
+		},
+		func(ctx context.Context, report *agent_entities.ChannelBalanceReport) bool {
+			if len(report.ChangedChannels) == 1 && report.UniqueId == "random_id" {
+				was_called = true
+			}
+
+			cancel()
+			return true
+		},
+	)
+
+	c.EventLoop()
+
+	select {
+	case <-time.After(5 * time.Second):
+		t.Fatal("Took too long")
+	case <-ctx.Done():
+		if !was_called {
+			t.Fatalf("Callback was not correctly invoked")
+		}
+	}
+}
+
+func TestBasicFlowFilterTwo(t *testing.T) {
+	pubKey, api, d := initTest(t)
+
+	d.HttpApi.DoFunc = func(req *http.Request) (*http.Response, error) {
+		contents := ""
+		if strings.Contains(req.URL.Path, "v1/getinfo") {
+			contents = getInfoJson("02b67e55fb850d7f7d77eb71038362bc0ed0abd5b7ee72cc4f90b16786c69b9256")
+		} else if strings.Contains(req.URL.Path, "v1/channels") {
+			contents = getChannelJson(1337, false, true)
+		}
+
+		r := ioutil.NopCloser(bytes.NewReader([]byte(contents)))
+
+		return &http.Response{
+			StatusCode: 200,
+			Body:       r,
+		}, nil
+	}
+
+	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(15*time.Second))
+
+	c := NewDefaultChannelChecker(ctx, time.Duration(0), true, false, nil)
+	// Make everything a bit faster
+	c.OverrideLoopInterval(1 * time.Second)
+	was_called := false
+
+	f, _ := filter.NewUnitTestFilter()
+	fd := f.(*filter.UnitTestFilter)
+	fd.AddAllowPubKey("02004c625d622245606a1ea2c1c69cfb4516b703b47945a3647713c05fe4aaeb1c")
+
+	c.Subscribe(
+		pubKey, "random_id",
+		func() lightning_api.LightingApiCalls { return api },
+		agent_entities.ReportingSettings{
+			AllowedEntropy:       64,
+			PollInterval:         agent_entities.SECOND,
+			AllowPrivateChannels: true,
+			Filter:               f,
+		},
+		func(ctx context.Context, report *agent_entities.ChannelBalanceReport) bool {
+			if len(report.ChangedChannels) == 2 && report.UniqueId == "random_id" {
+				was_called = true
+			}
+
+			cancel()
+			return true
+		},
+	)
+
+	c.EventLoop()
+
+	select {
+	case <-time.After(5 * time.Second):
+		t.Fatal("Took too long")
+	case <-ctx.Done():
+		if !was_called {
+			t.Fatalf("Callback was not correctly invoked")
+		}
+	}
+}
+
 func TestContextCanBeNil(t *testing.T) {
 	pubKey, api, d := initTest(t)