Skip to content

Commit

Permalink
Merge pull request #1452 from SamMayWork/query-api
Browse files Browse the repository at this point in the history
feat: Add an API to allow for querying events under a subscription with additional filtering
  • Loading branch information
nguyer authored Feb 12, 2024
2 parents fd542c0 + 45fc854 commit 662feec
Show file tree
Hide file tree
Showing 31 changed files with 1,988 additions and 401 deletions.
6 changes: 6 additions & 0 deletions docs/reference/config.md
Original file line number Diff line number Diff line change
Expand Up @@ -1390,6 +1390,12 @@ nav_order: 2
|batchSize|Default read ahead to enable for subscriptions that do not explicitly configure readahead|`int`|`50`
|batchTimeout|Default batch timeout|`int`|`50ms`

## subscription.events

|Key|Description|Type|Default Value|
|---|-----------|----|-------------|
|maxScanLength|The maximum number of events a search for historical events matching a subscription will index from the database|`int`|`1000`

## subscription.retry

|Key|Description|Type|Default Value|
Expand Down
393 changes: 393 additions & 0 deletions docs/swagger/swagger.yaml

Large diffs are not rendered by default.

123 changes: 121 additions & 2 deletions go.work.sum

Large diffs are not rendered by default.

74 changes: 74 additions & 0 deletions internal/apiserver/route_get_subscription_events_filtered.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
// Copyright © 2024 Kaleido, Inc.
//
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package apiserver

import (
"fmt"
"net/http"
"strconv"

"github.com/hyperledger/firefly-common/pkg/ffapi"
"github.com/hyperledger/firefly-common/pkg/i18n"
"github.com/hyperledger/firefly/internal/coremsgs"
"github.com/hyperledger/firefly/pkg/core"
"github.com/hyperledger/firefly/pkg/database"
)

var getSubscriptionEventsFiltered = &ffapi.Route{
Name: "getSubscriptionEventsFiltered",
Path: "subscriptions/{subid}/events",
Method: http.MethodGet,
PathParams: []*ffapi.PathParam{
{Name: "subid", Description: coremsgs.APIParamsSubscriptionID},
},
QueryParams: []*ffapi.QueryParam{
{Name: "startsequence", IsBool: false, Description: coremsgs.APISubscriptionStartSequenceID},
{Name: "endsequence", IsBool: false, Description: coremsgs.APISubscriptionEndSequenceID},
},
FilterFactory: database.EventQueryFactory,
Description: coremsgs.APIEndpointsGetSubscriptionEventsFiltered,
JSONInputValue: nil,
JSONOutputValue: func() interface{} { return []*core.Event{} },
JSONOutputCodes: []int{http.StatusOK},
Extensions: &coreExtensions{
CoreJSONHandler: func(r *ffapi.APIRequest, cr *coreRequest) (output interface{}, err error) {
subscription, _ := cr.or.GetSubscriptionByID(cr.ctx, r.PP["subid"])
var startSeq int
var endSeq int

if r.QP["startsequence"] != "" {
startSeq, err = strconv.Atoi(r.QP["startsequence"])
if err != nil {
return nil, i18n.NewError(cr.ctx, coremsgs.MsgSequenceIDDidNotParseToInt, fmt.Sprintf("startsequence: %s", r.QP["startsequence"]))
}
} else {
startSeq = -1
}

if r.QP["endsequence"] != "" {
endSeq, err = strconv.Atoi(r.QP["endsequence"])
if err != nil {
return nil, i18n.NewError(cr.ctx, coremsgs.MsgSequenceIDDidNotParseToInt, fmt.Sprintf("endsequence: %s", r.QP["endsequence"]))
}
} else {
endSeq = -1
}

return r.FilterResult(cr.or.GetSubscriptionEventsHistorical(cr.ctx, subscription, r.Filter, startSeq, endSeq))
},
},
}
85 changes: 85 additions & 0 deletions internal/apiserver/route_get_subscription_events_filtered_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
// Copyright © 2024 Kaleido, Inc.
//
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package apiserver

import (
"net/http/httptest"
"testing"

"github.com/hyperledger/firefly/pkg/core"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
)

func TestGetSubscriptionEventsFiltered(t *testing.T) {
o, r := newTestAPIServer()
o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/subscriptions/abcd12345/events?startsequence=100&endsequence=200", nil)
req.Header.Set("Content-Type", "application/json; charset=utf-8")
res := httptest.NewRecorder()

o.On("GetSubscriptionByID", mock.Anything, "abcd12345").
Return(&core.Subscription{}, nil)
o.On("GetSubscriptionEventsHistorical", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
Return([]*core.EnrichedEvent{}, nil, nil)

r.ServeHTTP(res, req)
assert.Equal(t, 200, res.Result().StatusCode)
}

func TestGetSubscriptionEventsFilteredStartSequenceIDDoesNotParse(t *testing.T) {
o, r := newTestAPIServer()
o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/subscriptions/abcd12345/events?startsequence=helloworld", nil)
req.Header.Set("Content-Type", "application/json; charset=utf-8")
res := httptest.NewRecorder()
o.On("GetSubscriptionByID", mock.Anything, "abcd12345").
Return(&core.Subscription{}, nil)

r.ServeHTTP(res, req)
assert.Equal(t, 400, res.Result().StatusCode)
assert.Contains(t, res.Body.String(), "helloworld")
}

func TestGetSubscriptionEventsFilteredEndSequenceIDDoesNotParse(t *testing.T) {
o, r := newTestAPIServer()
o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/subscriptions/abcd12345/events?endsequence=helloworld", nil)
req.Header.Set("Content-Type", "application/json; charset=utf-8")
res := httptest.NewRecorder()
o.On("GetSubscriptionByID", mock.Anything, "abcd12345").
Return(&core.Subscription{}, nil)

r.ServeHTTP(res, req)
assert.Equal(t, 400, res.Result().StatusCode)
assert.Contains(t, res.Body.String(), "helloworld")
}

func TestGetSubscriptionEventsFilteredNoSequenceIDsProvided(t *testing.T) {
o, r := newTestAPIServer()
o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/subscriptions/abcd12345/events", nil)
req.Header.Set("Content-Type", "application/json; charset=utf-8")
res := httptest.NewRecorder()
o.On("GetSubscriptionByID", mock.Anything, "abcd12345").
Return(&core.Subscription{}, nil)
o.On("GetSubscriptionEventsHistorical", mock.Anything, mock.Anything, mock.Anything, -1, -1).
Return([]*core.EnrichedEvent{}, nil, nil)

r.ServeHTTP(res, req)
assert.Equal(t, 200, res.Result().StatusCode)
}
3 changes: 2 additions & 1 deletion internal/apiserver/routes.go
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// Copyright © 2023 Kaleido, Inc.
// Copyright © 2024 Kaleido, Inc.
//
// SPDX-License-Identifier: Apache-2.0
//
Expand Down Expand Up @@ -110,6 +110,7 @@ var routes = append(
getStatusBatchManager,
getSubscriptionByID,
getSubscriptions,
getSubscriptionEventsFiltered,
getTokenAccountPools,
getTokenAccounts,
getTokenApprovals,
Expand Down
5 changes: 2 additions & 3 deletions internal/apiserver/server.go
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// Copyright © 2023 Kaleido, Inc.
// Copyright © 2024 Kaleido, Inc.
//
// SPDX-License-Identifier: Apache-2.0
//
Expand All @@ -23,8 +23,6 @@ import (
"strings"
"time"

"github.com/prometheus/client_golang/prometheus/promhttp"

"github.com/gorilla/mux"
"github.com/hyperledger/firefly-common/pkg/config"
"github.com/hyperledger/firefly-common/pkg/ffapi"
Expand All @@ -40,6 +38,7 @@ import (
"github.com/hyperledger/firefly/internal/metrics"
"github.com/hyperledger/firefly/internal/namespace"
"github.com/hyperledger/firefly/internal/orchestrator"
"github.com/prometheus/client_golang/prometheus/promhttp"
)

var (
Expand Down
3 changes: 3 additions & 0 deletions internal/coreconfig/coreconfig.go
Original file line number Diff line number Diff line change
Expand Up @@ -339,6 +339,8 @@ var (
SubscriptionsRetryMaxDelay = ffc("subscription.retry.maxDelay")
// SubscriptionsRetryFactor the backoff factor to use for retry of database operations
SubscriptionsRetryFactor = ffc("subscription.retry.factor")
// SubscriptionMaxHistoricalEventScanLength the maximum amount of historical events we scan for in the DB when indexing through old events against a subscription
SubscriptionMaxHistoricalEventScanLength = ffc("subscription.events.maxScanLength")
// TransactionWriterCount
TransactionWriterCount = ffc("transaction.writer.count")
// TransactionWriterBatchTimeout
Expand Down Expand Up @@ -459,6 +461,7 @@ func setDefaults() {
viper.SetDefault(string(SubscriptionsRetryInitialDelay), "250ms")
viper.SetDefault(string(SubscriptionsRetryMaxDelay), "30s")
viper.SetDefault(string(SubscriptionsRetryFactor), 2.0)
viper.SetDefault(string(SubscriptionMaxHistoricalEventScanLength), 1000)
viper.SetDefault(string(TransactionWriterBatchMaxTransactions), 100)
viper.SetDefault(string(TransactionWriterBatchTimeout), "10ms")
viper.SetDefault(string(TransactionWriterCount), 5)
Expand Down
6 changes: 5 additions & 1 deletion internal/coremsgs/en_api_translations.go
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// Copyright © 2023 Kaleido, Inc.
// Copyright © 2024 Kaleido, Inc.
//
// SPDX-License-Identifier: Apache-2.0
//
Expand Down Expand Up @@ -133,6 +133,7 @@ var (
APIEndpointsGetWebSockets = ffm("api.endpoints.getStatusWebSockets", "Gets a list of the current WebSocket connections to this node")
APIEndpointsGetStatus = ffm("api.endpoints.getStatus", "Gets the status of this namespace")
APIEndpointsGetSubscriptionByID = ffm("api.endpoints.getSubscriptionByID", "Gets a subscription by its ID")
APIEndpointsGetSubscriptionEventsFiltered = ffm("api.endpoints.getSubscriptionEventsFiltered", "Gets a collection of events filtered by the subscription for further filtering")
APIEndpointsGetSubscriptions = ffm("api.endpoints.getSubscriptions", "Gets a list of subscriptions")
APIEndpointsGetTokenAccountPools = ffm("api.endpoints.getTokenAccountPools", "Gets a list of token pools that contain a given token account key")
APIEndpointsGetTokenAccounts = ffm("api.endpoints.getTokenAccounts", "Gets a list of token accounts")
Expand Down Expand Up @@ -208,4 +209,7 @@ var (
APISmartContractDetails = ffm("api.smartContractDetails", "Additional smart contract details")
APISmartContractDetailsKey = ffm("api.smartContractDetailsKey", "Key")
APISmartContractDetailsValue = ffm("api.smartContractDetailsValue", "Value")

APISubscriptionStartSequenceID = ffm("api.startsequenceid", "The sequence ID in the raw event stream to start indexing through events from. Leave blank to start indexing from the most recent events")
APISubscriptionEndSequenceID = ffm("api.endsequenceid", "The sequence ID in the raw event stream to stop indexing through events at. Leave blank to start indexing from the most recent events")
)
7 changes: 4 additions & 3 deletions internal/coremsgs/en_config_descriptions.go
Original file line number Diff line number Diff line change
Expand Up @@ -383,9 +383,10 @@ var (
ConfigPluginSharedstorageIpfsGatewayURL = ffc("config.plugins.sharedstorage[].ipfs.gateway.url", "The URL for the IPFS Gateway", urlStringType)
ConfigPluginSharedstorageIpfsGatewayProxyURL = ffc("config.plugins.sharedstorage[].ipfs.gateway.proxy.url", "Optional HTTP proxy server to use when connecting to the IPFS Gateway", urlStringType)

ConfigSubscriptionMax = ffc("config.subscription.max", "The maximum number of pre-defined subscriptions that can exist (note for high fan-out consider connecting a dedicated pub/sub broker to the dispatcher)", i18n.IntType)
ConfigSubscriptionDefaultsBatchSize = ffc("config.subscription.defaults.batchSize", "Default read ahead to enable for subscriptions that do not explicitly configure readahead", i18n.IntType)
ConfigSubscriptionDefaultsBatchTimeout = ffc("config.subscription.defaults.batchTimeout", "Default batch timeout", i18n.IntType)
ConfigSubscriptionMax = ffc("config.subscription.max", "The maximum number of pre-defined subscriptions that can exist (note for high fan-out consider connecting a dedicated pub/sub broker to the dispatcher)", i18n.IntType)
ConfigSubscriptionDefaultsBatchSize = ffc("config.subscription.defaults.batchSize", "Default read ahead to enable for subscriptions that do not explicitly configure readahead", i18n.IntType)
ConfigSubscriptionDefaultsBatchTimeout = ffc("config.subscription.defaults.batchTimeout", "Default batch timeout", i18n.IntType)
ConfigSubscriptionMaxHistoricalEventScanLength = ffc("config.subscription.events.maxScanLength", "The maximum number of events a search for historical events matching a subscription will index from the database", i18n.IntType)

ConfigTokensName = ffc("config.tokens[].name", "A name to identify this token plugin", i18n.StringType)
ConfigTokensPlugin = ffc("config.tokens[].plugin", "The type of the token plugin to use", i18n.StringType)
Expand Down
Loading

0 comments on commit 662feec

Please sign in to comment.