Skip to content

Commit 14ebb04

Browse files
sreuland2opremiobarteknShaptic
authored
services/horizon, clients/horizonclient: Allow filtering ingested transactions by account or asset. (#4277)
Co-authored-by: Alfonso Acosta <[email protected]> Co-authored-by: Bartek Nowotarski <[email protected]> Co-authored-by: George <[email protected]>
1 parent 429ecee commit 14ebb04

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

64 files changed

+3086
-299
lines changed

‎clients/horizonclient/admin_client.go

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
package horizonclient
2+
3+
import (
4+
"bytes"
5+
"context"
6+
"encoding/json"
7+
"fmt"
8+
"net/http"
9+
"net/url"
10+
"time"
11+
12+
hProtocol "github.com/stellar/go/protocols/horizon"
13+
"github.com/stellar/go/support/errors"
14+
)
15+
16+
// port - the horizon admin port, zero value defaults to 4200
17+
// host - the host interface name that horizon has bound admin web service, zero value defaults to 'localhost'
18+
// timeout - the length of time for the http client to wait on responses from admin web service
19+
func NewAdminClient(port uint16, host string, timeout time.Duration) (*AdminClient, error) {
20+
baseURL, err := getAdminBaseURL(port, host)
21+
if err != nil {
22+
return nil, err
23+
}
24+
if timeout == 0 {
25+
timeout = HorizonTimeout
26+
}
27+
28+
return &AdminClient{
29+
baseURL: baseURL,
30+
http: http.DefaultClient,
31+
horizonTimeout: timeout,
32+
}, nil
33+
}
34+
35+
func getAdminBaseURL(port uint16, host string) (string, error) {
36+
baseURL, err := url.Parse("http://localhost")
37+
if err != nil {
38+
return "", err
39+
}
40+
adminPort := uint16(4200)
41+
if port > 0 {
42+
adminPort = port
43+
}
44+
adminHost := baseURL.Hostname()
45+
if len(host) > 0 {
46+
adminHost = host
47+
}
48+
baseURL.Host = fmt.Sprintf("%s:%d", adminHost, adminPort)
49+
return baseURL.String(), nil
50+
}
51+
52+
func (c *AdminClient) sendGetRequest(requestURL string, a interface{}) error {
53+
req, err := http.NewRequest("GET", requestURL, nil)
54+
if err != nil {
55+
return errors.Wrap(err, "error creating Admin HTTP request")
56+
}
57+
return c.sendHTTPRequest(req, a)
58+
}
59+
60+
func (c *AdminClient) sendHTTPRequest(req *http.Request, a interface{}) error {
61+
ctx, cancel := context.WithTimeout(context.Background(), c.horizonTimeout)
62+
defer cancel()
63+
64+
if resp, err := c.http.Do(req.WithContext(ctx)); err != nil {
65+
return err
66+
} else {
67+
return decodeResponse(resp, a, req.URL.String(), nil)
68+
}
69+
}
70+
71+
func (c *AdminClient) getIngestionFiltersURL(filter string) string {
72+
return fmt.Sprintf("%s/ingestion/filters/%s", c.baseURL, filter)
73+
}
74+
75+
func (c *AdminClient) GetIngestionAssetFilter() (hProtocol.AssetFilterConfig, error) {
76+
var filter hProtocol.AssetFilterConfig
77+
err := c.sendGetRequest(c.getIngestionFiltersURL("asset"), &filter)
78+
return filter, err
79+
}
80+
81+
func (c *AdminClient) GetIngestionAccountFilter() (hProtocol.AccountFilterConfig, error) {
82+
var filter hProtocol.AccountFilterConfig
83+
err := c.sendGetRequest(c.getIngestionFiltersURL("account"), &filter)
84+
return filter, err
85+
}
86+
87+
func (c *AdminClient) SetIngestionAssetFilter(filter hProtocol.AssetFilterConfig) error {
88+
buf := bytes.NewBuffer(nil)
89+
err := json.NewEncoder(buf).Encode(filter)
90+
if err != nil {
91+
return err
92+
}
93+
req, err := http.NewRequest(http.MethodPut, c.getIngestionFiltersURL("asset"), buf)
94+
if err != nil {
95+
return errors.Wrap(err, "error creating HTTP request")
96+
}
97+
req.Header.Add("Content-Type", "application/json")
98+
return c.sendHTTPRequest(req, nil)
99+
}
100+
101+
func (c *AdminClient) SetIngestionAccountFilter(filter hProtocol.AccountFilterConfig) error {
102+
buf := bytes.NewBuffer(nil)
103+
err := json.NewEncoder(buf).Encode(filter)
104+
if err != nil {
105+
return err
106+
}
107+
req, err := http.NewRequest(http.MethodPut, c.getIngestionFiltersURL("account"), buf)
108+
if err != nil {
109+
return errors.Wrap(err, "error creating HTTP request")
110+
}
111+
req.Header.Add("Content-Type", "application/json")
112+
return c.sendHTTPRequest(req, nil)
113+
}
114+
115+
// ensure that the horizon admin client implements AdminClientInterface
116+
var _ AdminClientInterface = &AdminClient{}

‎clients/horizonclient/admin_client_test.go

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
package horizonclient
2+
3+
import (
4+
"testing"
5+
6+
"github.com/stretchr/testify/assert"
7+
"github.com/stretchr/testify/require"
8+
)
9+
10+
func TestDefaultAdminHostPort(t *testing.T) {
11+
horizonAdminClient, err := NewAdminClient(0, "", 0)
12+
13+
fullAdminURL := horizonAdminClient.getIngestionFiltersURL("test")
14+
require.NoError(t, err)
15+
assert.Equal(t, "http://localhost:4200/ingestion/filters/test", fullAdminURL)
16+
}
17+
18+
func TestOverrideAdminHostPort(t *testing.T) {
19+
horizonAdminClient, err := NewAdminClient(1234, "127.0.0.1", 0)
20+
21+
fullAdminURL := horizonAdminClient.getIngestionFiltersURL("test")
22+
require.NoError(t, err)
23+
assert.Equal(t, "http://127.0.0.1:1234/ingestion/filters/test", fullAdminURL)
24+
}

‎clients/horizonclient/client.go

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,7 @@ func (c *Client) sendHTTPRequest(req *http.Request, a interface{}) error {
122122
if resp, err := c.HTTP.Do(req.WithContext(ctx)); err != nil {
123123
return err
124124
} else {
125-
return decodeResponse(resp, &a, c)
125+
return decodeResponse(resp, a, c.HorizonURL, c.clock)
126126
}
127127
}
128128

@@ -270,6 +270,9 @@ func (c *Client) setDefaultClient() {
270270
// fixHorizonURL strips all slashes(/) at the end of HorizonURL if any, then adds a single slash
271271
func (c *Client) fixHorizonURL() string {
272272
c.fixHorizonURLOnce.Do(func() {
273+
// TODO: we shouldn't happily edit data provided by the user,
274+
// better store it in an internal variable or, even better,
275+
// just parse it every time (what if the url changes during the life of the client?).
273276
c.HorizonURL = strings.TrimRight(c.HorizonURL, "/") + "/"
274277
})
275278
return c.HorizonURL

‎clients/horizonclient/internal.go

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -8,19 +8,24 @@ import (
88
"strings"
99
"time"
1010

11+
"github.com/stellar/go/support/clock"
1112
"github.com/stellar/go/support/errors"
1213
)
1314

1415
// decodeResponse decodes the response from a request to a horizon server
15-
func decodeResponse(resp *http.Response, object interface{}, hc *Client) (err error) {
16+
func decodeResponse(resp *http.Response, object interface{}, horizonUrl string, clock *clock.Clock) (err error) {
1617
defer resp.Body.Close()
18+
if object == nil {
19+
// Nothing to decode
20+
return nil
21+
}
1722
decoder := json.NewDecoder(resp.Body)
1823

19-
u, err := url.Parse(hc.HorizonURL)
24+
u, err := url.Parse(horizonUrl)
2025
if err != nil {
21-
return errors.Errorf("unable to parse the provided horizon url: %s", hc.HorizonURL)
26+
return errors.Errorf("unable to parse the provided horizon url: %s", horizonUrl)
2227
}
23-
setCurrentServerTime(u.Hostname(), resp.Header["Date"], hc)
28+
setCurrentServerTime(u.Hostname(), resp.Header["Date"], clock)
2429

2530
if !(resp.StatusCode >= 200 && resp.StatusCode < 300) {
2631
horizonError := &Error{
@@ -32,7 +37,6 @@ func decodeResponse(resp *http.Response, object interface{}, hc *Client) (err er
3237
}
3338
return horizonError
3439
}
35-
3640
err = decoder.Decode(&object)
3741
if err != nil {
3842
return errors.Wrap(err, "error decoding response")
@@ -120,7 +124,7 @@ func addQueryParams(params ...interface{}) string {
120124
}
121125

122126
// setCurrentServerTime saves the current time returned by a horizon server
123-
func setCurrentServerTime(host string, serverDate []string, hc *Client) {
127+
func setCurrentServerTime(host string, serverDate []string, clock *clock.Clock) {
124128
if len(serverDate) == 0 {
125129
return
126130
}
@@ -129,7 +133,7 @@ func setCurrentServerTime(host string, serverDate []string, hc *Client) {
129133
return
130134
}
131135
serverTimeMapMutex.Lock()
132-
ServerTimeMap[host] = ServerTimeRecord{ServerTime: st.UTC().Unix(), LocalTimeRecorded: hc.clock.Now().UTC().Unix()}
136+
ServerTimeMap[host] = ServerTimeRecord{ServerTime: st.UTC().Unix(), LocalTimeRecorded: clock.Now().UTC().Unix()}
133137
serverTimeMapMutex.Unlock()
134138
}
135139

‎clients/horizonclient/main.go

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -149,11 +149,29 @@ type Client struct {
149149
clock *clock.Clock
150150
}
151151

152+
type AdminClient struct {
153+
// fully qualified url for the admin web service
154+
baseURL string
155+
156+
// HTTP client to make requests with
157+
http HTTP
158+
159+
// max client wait time for response
160+
horizonTimeout time.Duration
161+
}
162+
152163
// SubmitTxOpts represents the submit transaction options
153164
type SubmitTxOpts struct {
154165
SkipMemoRequiredCheck bool
155166
}
156167

168+
type AdminClientInterface interface {
169+
GetIngestionAccountFilter() (hProtocol.AccountFilterConfig, error)
170+
GetIngestionAssetFilter() (hProtocol.AssetFilterConfig, error)
171+
SetIngestionAccountFilter(hProtocol.AccountFilterConfig) error
172+
SetIngestionAssetFilter(hProtocol.AssetFilterConfig) error
173+
}
174+
157175
// ClientInterface contains methods implemented by the horizon client
158176
type ClientInterface interface {
159177
Accounts(request AccountsRequest) (hProtocol.AccountsPage, error)

‎clients/horizonclient/mocks.go

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,10 @@ type MockClient struct {
1515
mock.Mock
1616
}
1717

18+
type MockAdminClient struct {
19+
mock.Mock
20+
}
21+
1822
// Accounts is a mocking method
1923
func (m *MockClient) Accounts(request AccountsRequest) (hProtocol.AccountsPage, error) {
2024
a := m.Called(request)
@@ -349,5 +353,28 @@ func (m *MockClient) PrevLiquidityPoolsPage(page hProtocol.LiquidityPoolsPage) (
349353
return a.Get(0).(hProtocol.LiquidityPoolsPage), a.Error(1)
350354
}
351355

356+
func (m *MockAdminClient) GetIngestionAccountFilter() (hProtocol.AccountFilterConfig, error) {
357+
a := m.Called()
358+
return a.Get(0).(hProtocol.AccountFilterConfig), a.Error(1)
359+
}
360+
361+
func (m *MockAdminClient) GetIngestionAssetFilter() (hProtocol.AssetFilterConfig, error) {
362+
a := m.Called()
363+
return a.Get(0).(hProtocol.AssetFilterConfig), a.Error(1)
364+
}
365+
366+
func (m *MockAdminClient) SetIngestionAccountFilter(resource hProtocol.AccountFilterConfig) error {
367+
a := m.Called(resource)
368+
return a.Error(0)
369+
}
370+
371+
func (m *MockAdminClient) SetIngestionAssetFilter(resource hProtocol.AssetFilterConfig) error {
372+
a := m.Called(resource)
373+
return a.Error(0)
374+
}
375+
352376
// ensure that the MockClient implements ClientInterface
353377
var _ ClientInterface = &MockClient{}
378+
379+
// ensure that the MockClient implements ClientInterface
380+
var _ AdminClientInterface = &MockAdminClient{}

‎protocols/horizon/main.go

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -843,3 +843,55 @@ type LiquidityPoolReserve struct {
843843
Asset string `json:"asset"`
844844
Amount string `json:"amount"`
845845
}
846+
847+
type AssetFilterConfig struct {
848+
Whitelist []string `json:"whitelist"`
849+
Enabled *bool `json:"enabled"`
850+
LastModified int64 `json:"last_modified,omitempty"`
851+
}
852+
853+
type AccountFilterConfig struct {
854+
Whitelist []string `json:"whitelist"`
855+
Enabled *bool `json:"enabled"`
856+
LastModified int64 `json:"last_modified,omitempty"`
857+
}
858+
859+
func (f *AccountFilterConfig) UnmarshalJSON(data []byte) error {
860+
type accountFilterConfig AccountFilterConfig
861+
var config = accountFilterConfig{}
862+
863+
if err := json.Unmarshal(data, &config); err != nil {
864+
return err
865+
}
866+
867+
if config.Whitelist == nil {
868+
return errors.New("missing required whitelist")
869+
}
870+
871+
if config.Enabled == nil {
872+
return errors.New("missing required enabled")
873+
}
874+
875+
*f = AccountFilterConfig(config)
876+
return nil
877+
}
878+
879+
func (f *AssetFilterConfig) UnmarshalJSON(data []byte) error {
880+
type assetFilterConfig AssetFilterConfig
881+
var config = assetFilterConfig{}
882+
883+
if err := json.Unmarshal(data, &config); err != nil {
884+
return err
885+
}
886+
887+
if config.Whitelist == nil {
888+
return errors.New("missing required whitelist")
889+
}
890+
891+
if config.Enabled == nil {
892+
return errors.New("missing required enabled")
893+
}
894+
895+
*f = AssetFilterConfig(config)
896+
return nil
897+
}

‎services/horizon/CHANGELOG.md

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,19 @@ This is the final release after the [release candidate](v2.17.0-release-candidat
1313

1414
- Timebounds within the `preconditions` object are strings containing int64 UNIX timestamps in seconds rather than formatted date-times (which was a bug) ([4361](https://github.com/stellar/go/pull/4361)).
1515

16+
* New Ingestion Filters Feature: Provide the ability to select which ledger transactions are accepted at ingestion time to be stored on horizon's historical databse.
17+
18+
Define filter rules through Admin API and the historical ingestion process will check the rules and only persist the ledger transactions that pass the filter rules. Initially, two filters and corresponding rules are possible:
19+
20+
* 'whitelist by account id' ([4221](https://github.com/stellar/go/issues/4221))
21+
* 'whitelist by canonical asset id' ([4222](https://github.com/stellar/go/issues/4222))
22+
23+
The filters and their configuration are optional features and must be enabled with horizon command line parameters `admin-port=4200` and `enable-ingestion-filtering=true`
24+
25+
Once set, filter configurations and their rules are initially empty and the filters are disabled by default. To enable filters, update the configuration settings, refer to the Admin API Docs which are published on the Admin Port at http://localhost:<admin_port>/, follow details and examples for endpoints:
26+
* `/ingestion/filters/account`
27+
* `/ingestion/filters/asset.`
28+
1629
## V2.17.0 Release Candidate
1730

1831
**Upgrading to this version from <= v2.8.3 will trigger a state rebuild. During this process (which will take at least 10 minutes), Horizon will not ingest new ledgers.**

‎services/horizon/cmd/db.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -384,6 +384,7 @@ func runDBReingestRange(ledgerRanges []history.LedgerRange, reingestForce bool,
384384
NetworkPassphrase: config.NetworkPassphrase,
385385
HistoryArchiveURL: config.HistoryArchiveURLs[0],
386386
CheckpointFrequency: config.CheckpointFrequency,
387+
ReingestEnabled: true,
387388
MaxReingestRetries: int(retries),
388389
ReingestRetryBackoffSeconds: int(retryBackoffSeconds),
389390
EnableCaptiveCore: config.EnableCaptiveCoreIngestion,
@@ -395,6 +396,7 @@ func runDBReingestRange(ledgerRanges []history.LedgerRange, reingestForce bool,
395396
StellarCoreCursor: config.CursorName,
396397
StellarCoreURL: config.StellarCoreURL,
397398
RoundingSlippageFilter: config.RoundingSlippageFilter,
399+
EnableIngestionFiltering: config.EnableIngestionFiltering,
398400
}
399401

400402
if ingestConfig.HistorySession, err = db.Open("postgres", config.DatabaseURL); err != nil {

0 commit comments

Comments
 (0)