From e7d03bf1117e7caf203db19db5d9dd477260aa7b Mon Sep 17 00:00:00 2001 From: bubbajoe Date: Mon, 17 Jun 2024 23:54:41 +0900 Subject: [PATCH] improve extractor promise wait timeout add prevotereq to rafthttp optimize promise waiting change example domains so they work on raft instances add admin test --- README.md | 2 +- TODO.md | 17 +- config.dgate.yaml | 7 +- functional-tests/admin_tests/admin_test.sh | 41 +++ functional-tests/admin_tests/admin_test.ts | 4 + .../admin_tests/change_checker.sh | 6 +- .../admin_tests/iphash_load_balancer_test.sh | 6 +- .../admin_tests/merge_responses_test.sh | 4 +- .../admin_tests/modify_request_test.sh | 4 +- .../admin_tests/modify_response_test.sh | 4 +- .../admin_tests/multi_module_test.sh | 6 +- .../admin_tests/performance_test_prep.sh | 8 +- .../admin_tests/url_shortener_test.sh | 12 +- functional-tests/raft_tests/raft_test.sh | 28 +- go.mod | 22 +- go.sum | 56 ++-- internal/admin/admin_fsm.go | 56 ++-- internal/admin/admin_raft.go | 69 +++-- internal/admin/changestate/change_state.go | 6 +- .../changestate/testutil/change_state.go | 6 +- internal/admin/routes/misc_routes.go | 71 +++-- internal/config/config.go | 3 +- internal/config/configtest/dgate_configs.go | 2 +- internal/config/loader.go | 1 + internal/config/store_config.go | 5 +- internal/proxy/change_log.go | 250 ++++++++++-------- internal/proxy/dynamic_proxy.go | 146 +++++----- internal/proxy/proxy_handler.go | 4 +- internal/proxy/proxy_printer.go | 43 +-- internal/proxy/proxy_state.go | 169 ++++++------ internal/proxy/proxystore/proxy_store.go | 15 +- internal/proxy/runtime_context.go | 8 +- internal/proxy/util.go | 12 +- performance-tests/long-perf-test.js | 2 +- performance-tests/perf-test.js | 2 +- pkg/modules/extractors/async_tracker.go | 66 ----- pkg/modules/extractors/extractors.go | 39 ++- pkg/modules/extractors/extractors_test.go | 7 +- pkg/modules/extractors/runtime_test.go | 3 +- pkg/modules/testutil/testutil.go | 2 +- pkg/rafthttp/rafthttp.go | 8 + pkg/storage/debug_storage.go | 93 ------- pkg/storage/file_storage.go | 214 ++++++--------- pkg/storage/mem_storage.go | 133 ++++++++++ pkg/storage/memory_storage.go | 27 -- pkg/typescript/typescript.go | 8 +- pkg/typescript/typescript_test.go | 3 +- pkg/util/http_test.go | 22 +- pkg/util/parse.go | 14 +- pkg/util/parse_test.go | 20 ++ 50 files changed, 934 insertions(+), 822 deletions(-) create mode 100755 functional-tests/admin_tests/admin_test.sh create mode 100644 functional-tests/admin_tests/admin_test.ts delete mode 100644 pkg/modules/extractors/async_tracker.go delete mode 100644 pkg/storage/debug_storage.go create mode 100644 pkg/storage/mem_storage.go delete mode 100644 pkg/storage/memory_storage.go create mode 100644 pkg/util/parse_test.go diff --git a/README.md b/README.md index de75246..2617067 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ DGate is a distributed API Gateway built for developers. DGate allows you to use ## Getting Started -Coming soon @ http://dgate.io/docs/getting-started +http://dgate.io/docs/getting-started ### Installing diff --git a/TODO.md b/TODO.md index 5bb5d29..9760e6d 100644 --- a/TODO.md +++ b/TODO.md @@ -8,6 +8,12 @@ - cluster management (raft commands, replica commands, etc.) (low priority) - other commands (backup, restore, etc.) (low priority) +# Raft Snapshots + +- Add support for Raft snapshots to reduce the size of the Raft log. This can be used to reduce the size of the Raft log and improve the performance of the cluster. + - [ ] - Snapshot documents + - [ ] - Snapshot resources (with correlactions) + ## Add Module Tests - Test multiple modules being used at the same time @@ -138,13 +144,12 @@ Make it easier to debug modules by adding more logging and error handling. This Add stack tracing for typescript modules. - -## Decouple Admin API from Raft Implementation - -Currently, Raft Implementation is tightly coupled with the Admin API. This makes it difficult to change the Raft Implementation without changing the Admin API. Decouple the Raft Implementation from the Admin API to make it easier to change the Raft Implementation. - ## Add Telemetry (sentry, datadog, etc.) ## ResourceManager callback for resource changes -Add a callback to the ResourceManager that is called when a resource is changed. This can be used to invalidate caches, update modules, and more. \ No newline at end of file +Add a callback to the ResourceManager that is called when a resource is changed. This can be used to invalidate caches, update modules, and more. + +## Enable WAF + +https://github.com/corazawaf/coraza diff --git a/config.dgate.yaml b/config.dgate.yaml index aeeca0d..d3fa32b 100644 --- a/config.dgate.yaml +++ b/config.dgate.yaml @@ -1,9 +1,8 @@ version: v1 debug: true -log_level: ${LOG_LEVEL:-info} +log_level: ${LOG_LEVEL:-debug} disable_default_namespace: true -tags: - - debug +tags: [debug, local, test] storage: type: file dir: .dgate/data/ @@ -15,7 +14,7 @@ test_server: proxy: port: ${PORT:-80} host: 0.0.0.0 - enable_console_logger: true + console_log_level: info transport: dns_prefer_go: true init_resources: diff --git a/functional-tests/admin_tests/admin_test.sh b/functional-tests/admin_tests/admin_test.sh new file mode 100755 index 0000000..3cd62e0 --- /dev/null +++ b/functional-tests/admin_tests/admin_test.sh @@ -0,0 +1,41 @@ +#!/bin/bash + +set -eo xtrace + +ADMIN_URL=${ADMIN_URL:-"http://localhost:9080"} +PROXY_URL=${PROXY_URL:-"http://localhost"} + +DIR="$( cd "$( dirname "$0" )" && pwd )" + +# domain setup + +id=$(uuid) + +dgate-cli namespace create name=ns-$id + +dgate-cli domain create name=dm-$id \ + namespace=ns-$id priority:=$RANDOM patterns="$id.example.com" + +dgate-cli service create \ + name=svc-$id namespace=ns-$id \ + urls="http://localhost:8888/$RANDOM" + +dgate-cli module create name=module1 \ + payload@=$DIR/admin_test.ts \ + namespace=ns-$id + +dgate-cli route create \ + name=rt-$id \ + service=svc-$id \ + namespace=ns-$id \ + paths="/,/{},/$id,/$id/{id}" \ + methods=GET,POST,PUT \ + modules=module1 \ + preserveHost:=false \ + stripPath:=false + +curl -f $ADMIN_URL/readyz + +curl -f ${PROXY_URL}/$id/$RANDOM-$j -H Host:$id.example.com + +echo "Admin Test Succeeded" diff --git a/functional-tests/admin_tests/admin_test.ts b/functional-tests/admin_tests/admin_test.ts new file mode 100644 index 0000000..eac6297 --- /dev/null +++ b/functional-tests/admin_tests/admin_test.ts @@ -0,0 +1,4 @@ + +export const responseModifier = async (ctx: any) => { + console.log("responseModifier -> path params", ctx.pathParams()); +} \ No newline at end of file diff --git a/functional-tests/admin_tests/change_checker.sh b/functional-tests/admin_tests/change_checker.sh index 84dd271..18fe1af 100755 --- a/functional-tests/admin_tests/change_checker.sh +++ b/functional-tests/admin_tests/change_checker.sh @@ -14,7 +14,7 @@ dgate-cli namespace create \ dgate-cli domain create \ name=change_checker-dm \ - patterns:='["change_checker.com"]' \ + patterns:='["change_checker.example.com"]' \ namespace=change_checker-ns dgate-cli module create name=change_checker-mod \ @@ -29,7 +29,7 @@ dgate-cli route create \ preserveHost:=true \ namespace=change_checker-ns -MODID1=$(curl -sG -H Host:change_checker.com ${PROXY_URL}/ | jq -r '.mod') +MODID1=$(curl -sG -H Host:change_checker.example.com ${PROXY_URL}/ | jq -r '.mod') if [ "$MODID1" != "module1" ]; then echo "Initial assert failed" @@ -43,7 +43,7 @@ dgate-cli module create name=change_checker-mod \ # dgate-cli r.ker-ns -MODID2=$(curl -sG -H Host:change_checker.com ${PROXY_URL}/ | jq -r '.mod') +MODID2=$(curl -sG -H Host:change_checker.example.com ${PROXY_URL}/ | jq -r '.mod') if [ "$MODID2" != "module2" ]; then echo "module update failed" diff --git a/functional-tests/admin_tests/iphash_load_balancer_test.sh b/functional-tests/admin_tests/iphash_load_balancer_test.sh index 4a716e1..b30b5cc 100755 --- a/functional-tests/admin_tests/iphash_load_balancer_test.sh +++ b/functional-tests/admin_tests/iphash_load_balancer_test.sh @@ -14,7 +14,7 @@ dgate-cli namespace create \ dgate-cli domain create \ name=test-lb-dm \ - patterns:='["test-lb.com"]' \ + patterns:='["test-lb.example.com"]' \ namespace=test-lb-ns MOD_B64="$(base64 < $DIR/iphash_load_balancer.ts)" @@ -39,9 +39,9 @@ dgate-cli route create \ preserveHost:=true \ namespace=test-lb-ns -path1="$(curl -s --fail-with-body ${PROXY_URL}/test-lb -H Host:test-lb.com | jq -r '.data.path')" +path1="$(curl -s --fail-with-body ${PROXY_URL}/test-lb -H Host:test-lb.example.com | jq -r '.data.path')" -path2="$(curl -s --fail-with-body ${PROXY_URL}/test-lb -H Host:test-lb.com -H X-Forwarded-For:192.168.0.1 | jq -r '.data.path')" +path2="$(curl -s --fail-with-body ${PROXY_URL}/test-lb -H Host:test-lb.example.com -H X-Forwarded-For:192.168.0.1 | jq -r '.data.path')" if [ "$path1" != "$path2" ]; then echo "IP Hash Load Balancer Test Passed" diff --git a/functional-tests/admin_tests/merge_responses_test.sh b/functional-tests/admin_tests/merge_responses_test.sh index 83752c9..126d45b 100755 --- a/functional-tests/admin_tests/merge_responses_test.sh +++ b/functional-tests/admin_tests/merge_responses_test.sh @@ -14,7 +14,7 @@ dgate-cli namespace create \ dgate-cli domain create \ name=test-dm \ - patterns:='["test.com"]' \ + patterns:='["test.example.com"]' \ namespace=test-ns MOD_B64="$(base64 < $DIR/merge_responses.ts)" @@ -32,6 +32,6 @@ dgate-cli route create \ preserveHost:=true \ namespace=test-ns -curl -s --fail-with-body ${PROXY_URL}/hello -H Host:test.com +curl -s --fail-with-body ${PROXY_URL}/hello -H Host:test.example.com echo "Merge Responses Test Passed" diff --git a/functional-tests/admin_tests/modify_request_test.sh b/functional-tests/admin_tests/modify_request_test.sh index df42d71..8f25ca9 100755 --- a/functional-tests/admin_tests/modify_request_test.sh +++ b/functional-tests/admin_tests/modify_request_test.sh @@ -14,7 +14,7 @@ dgate-cli namespace create \ dgate-cli domain create \ name=modify_request_test-dm \ - patterns:='["modify_request_test.com"]' \ + patterns:='["modify_request_test.example.com"]' \ namespace=modify_request_test-ns MOD_B64="$(base64 < $DIR/modify_request.ts)" @@ -38,7 +38,7 @@ dgate-cli route create \ service='base_svc' curl -s --fail-with-body ${PROXY_URL}/modify_request_test \ - -H Host:modify_request_test.com \ + -H Host:modify_request_test.example.com \ -H X-Forwarded-For:1.1.1.1 echo "Modify Request Test Passed" diff --git a/functional-tests/admin_tests/modify_response_test.sh b/functional-tests/admin_tests/modify_response_test.sh index 8d8196c..90f98fc 100755 --- a/functional-tests/admin_tests/modify_response_test.sh +++ b/functional-tests/admin_tests/modify_response_test.sh @@ -14,7 +14,7 @@ dgate-cli namespace create \ dgate-cli domain create \ name=test-dm \ - patterns:='["test.com"]' \ + patterns:='["test.example.com"]' \ namespace=test-ns MOD_B64="$(base64 < $DIR/modify_response.ts)" @@ -37,6 +37,6 @@ dgate-cli route create \ namespace=test-ns \ service='base_svc' -curl -s ${PROXY_URL}/test -H Host:test.com +curl -s ${PROXY_URL}/test -H Host:test.example.com echo "Modify Response Test Passed" diff --git a/functional-tests/admin_tests/multi_module_test.sh b/functional-tests/admin_tests/multi_module_test.sh index f5071f9..9d69f4a 100755 --- a/functional-tests/admin_tests/multi_module_test.sh +++ b/functional-tests/admin_tests/multi_module_test.sh @@ -14,7 +14,7 @@ dgate-cli namespace create \ dgate-cli domain create \ name=multimod-test-dm \ - patterns:='["multimod-test.com"]' \ + patterns:='["multimod-test.example.com"]' \ namespace=multimod-test-ns MOD_B64=$(base64 <<-END @@ -71,7 +71,7 @@ dgate-cli route create name=base_rt \ namespace=multimod-test-ns -curl -s --fail-with-body ${PROXY_URL}/ -H Host:multimod-test.com -curl -s --fail-with-body ${PROXY_URL}/multimod-test -H Host:multimod-test.com +curl -s --fail-with-body ${PROXY_URL}/ -H Host:multimod-test.example.com +curl -s --fail-with-body ${PROXY_URL}/multimod-test -H Host:multimod-test.example.com echo "Multi Module Test Passed" \ No newline at end of file diff --git a/functional-tests/admin_tests/performance_test_prep.sh b/functional-tests/admin_tests/performance_test_prep.sh index a9a8f3e..2ed9284 100755 --- a/functional-tests/admin_tests/performance_test_prep.sh +++ b/functional-tests/admin_tests/performance_test_prep.sh @@ -14,7 +14,7 @@ dgate-cli -V -f namespace create \ name=test-ns1 dgate-cli domain create \ - name=test-dm patterns:='["dgate.dev"]' \ + name=test-dm patterns:='["performance.example.com"]' \ namespace=test-ns1 priority:=100 dgate-cli service create \ @@ -53,10 +53,10 @@ dgate-cli route create \ namespace=test-ns1 -curl -s --fail-with-body ${PROXY_URL}/svctest -H Host:dgate.dev +curl -s --fail-with-body ${PROXY_URL}/svctest -H Host:performance.example.com -curl -s --fail-with-body ${PROXY_URL}/modtest -H Host:dgate.dev +curl -s --fail-with-body ${PROXY_URL}/modtest -H Host:performance.example.com -curl -s ${PROXY_URL}/blank -H Host:dgate.dev +curl -s ${PROXY_URL}/blank -H Host:performance.example.com echo "Performance Test Prep Done" \ No newline at end of file diff --git a/functional-tests/admin_tests/url_shortener_test.sh b/functional-tests/admin_tests/url_shortener_test.sh index fcbf135..973ed13 100755 --- a/functional-tests/admin_tests/url_shortener_test.sh +++ b/functional-tests/admin_tests/url_shortener_test.sh @@ -14,14 +14,12 @@ dgate-cli namespace create \ dgate-cli domain create \ name=url_shortener-dm \ - patterns:='["url_shortener.com"]' \ + patterns:='["url_shortener.example.com"]' \ namespace=url_shortener-ns dgate-cli collection create \ schema:='{"type":"object","properties":{"url":{"type":"string"}}}' \ - name=short_link \ - type=document \ - namespace=url_shortener-ns + name=short_link type=document namespace=url_shortener-ns dgate-cli module create name=url_shortener-mod \ payload@=$DIR/url_shortener.ts \ @@ -36,11 +34,11 @@ dgate-cli route create \ namespace=url_shortener-ns JSON_RESP=$(curl -sG -X POST \ - -H Host:url_shortener.com ${PROXY_URL}/ \ - --data-urlencode 'url=https://dgate.io') + -H Host:url_shortener.example.com ${PROXY_URL}/ \ + --data-urlencode 'url=https://dgate.io/'$(uuid)) URL_ID=$(echo $JSON_RESP | jq -r '.id') curl -s --fail-with-body \ ${PROXY_URL}/$URL_ID \ - -H Host:url_shortener.com + -H Host:url_shortener.example.com diff --git a/functional-tests/raft_tests/raft_test.sh b/functional-tests/raft_tests/raft_test.sh index 7829b1b..c0a894b 100755 --- a/functional-tests/raft_tests/raft_test.sh +++ b/functional-tests/raft_tests/raft_test.sh @@ -40,30 +40,30 @@ dgate-cli -f route create \ name=rt-$id \ service=svc-$id \ namespace=ns-$id \ - paths="/$id/{id}" \ - methods:='["GET"]' \ + paths="/,/{},/$id,/$id/{id}" \ + methods=GET,POST,PUT \ preserveHost:=false \ - stripPath:=true + stripPath:=false curl -f $ADMIN_URL1/readyz -for i in {1..5}; do +for i in {1..1}; do for j in {1..3}; do proxy_url=PROXY_URL$i - curl -f ${!proxy_url}/$id/$j -H Host:$id.example.com + curl -f ${!proxy_url}/$id/$RANDOM-$j -H Host:$id.example.com done done -if dgate-cli --admin $ADMIN_URL4 namespace create name=0; then - echo "Expected error when creating namespace on non-voter" - exit 1 -fi +# if dgate-cli --admin $ADMIN_URL4 namespace create name=0; then +# echo "Expected error when creating namespace on non-voter" +# exit 1 +# fi -export DGATE_ADMIN_API=$ADMIN_URL5 +# export DGATE_ADMIN_API=$ADMIN_URL5 -if dgate-cli --admin $ADMIN_URL5 namespace create name=0; then - echo "Expected error when creating namespace on non-voter" - exit 1 -fi +# if dgate-cli --admin $ADMIN_URL5 namespace create name=0; then +# echo "Expected error when creating namespace on non-voter" +# exit 1 +# fi echo "Raft Test Succeeded" diff --git a/go.mod b/go.mod index 772cf23..b77eff7 100644 --- a/go.mod +++ b/go.mod @@ -5,13 +5,13 @@ go 1.22.0 require ( github.com/clarkmcc/go-typescript v0.7.0 github.com/dgate-io/chi-router v0.0.0-20231217131951-d154152d5115 - github.com/dgate-io/raft-badger v0.0.0-20231217131807-c5eb3f9eafa5 github.com/dgraph-io/badger/v4 v4.2.0 github.com/dop251/goja v0.0.0-20240220182346-e401ed450204 github.com/dop251/goja_nodejs v0.0.0-20231122114759-e84d9a924c5c github.com/google/uuid v1.3.1 - github.com/hashicorp/go-hclog v1.6.2 - github.com/hashicorp/raft v1.6.0 + github.com/hashicorp/go-hclog v1.6.3 + github.com/hashicorp/raft v1.7.0 + github.com/hashicorp/raft-boltdb/v2 v2.3.0 github.com/knadh/koanf/parsers/json v0.1.0 github.com/knadh/koanf/parsers/toml v0.1.0 github.com/knadh/koanf/parsers/yaml v0.1.0 @@ -38,14 +38,16 @@ require ( require ( github.com/armon/go-metrics v0.4.1 // indirect github.com/beorn7/perks v1.0.1 // indirect + github.com/boltdb/bolt v1.3.1 // indirect github.com/cespare/xxhash/v2 v2.2.0 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/dgraph-io/ristretto v0.1.1 // indirect + github.com/dgryski/go-farm v0.0.0-20191112170834-c2139c5d712b // indirect github.com/dlclark/regexp2 v1.10.0 // indirect github.com/dop251/base64dec v0.0.0-20231022112746-c6c9f9a96217 // indirect github.com/dustin/go-humanize v1.0.0 // indirect - github.com/fatih/color v1.14.1 // indirect + github.com/fatih/color v1.17.0 // indirect github.com/fsnotify/fsnotify v1.7.0 // indirect github.com/go-logr/logr v1.4.1 // indirect github.com/go-logr/stdr v1.2.2 // indirect @@ -58,12 +60,14 @@ require ( github.com/google/flatbuffers v1.12.1 // indirect github.com/google/pprof v0.0.0-20230926050212-f7f687d19a98 // indirect github.com/hashicorp/go-immutable-radix v1.3.1 // indirect - github.com/hashicorp/go-msgpack/v2 v2.1.1 // indirect - github.com/hashicorp/golang-lru v0.5.4 // indirect + github.com/hashicorp/go-msgpack v1.1.5 // indirect + github.com/hashicorp/go-msgpack/v2 v2.1.2 // indirect + github.com/hashicorp/golang-lru v1.0.2 // indirect + github.com/hashicorp/raft-boltdb v0.0.0-20231211162105-6c830fa4535e // indirect github.com/klauspost/compress v1.17.0 // indirect github.com/knadh/koanf/maps v0.1.1 // indirect github.com/mattn/go-colorable v0.1.13 // indirect - github.com/mattn/go-isatty v0.0.19 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect github.com/pelletier/go-toml v1.9.5 // indirect @@ -76,11 +80,13 @@ require ( github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/stretchr/objx v0.5.2 // indirect github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect + go.etcd.io/bbolt v1.3.10 // indirect go.opencensus.io v0.24.0 // indirect go.opentelemetry.io/otel/sdk v1.26.0 // indirect go.opentelemetry.io/otel/trace v1.26.0 // indirect go.uber.org/multierr v1.11.0 // indirect - golang.org/x/sys v0.19.0 // indirect + golang.org/x/sync v0.7.0 // indirect + golang.org/x/sys v0.21.0 // indirect golang.org/x/text v0.14.0 // indirect google.golang.org/protobuf v1.33.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 7bb3c79..1a1907c 100644 --- a/go.sum +++ b/go.sum @@ -1,16 +1,21 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= +github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ= github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= +github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878/go.mod h1:3AMJUQhVx52RsWOnlkpikZr01T/yAVN2gn0861vByNg= +github.com/armon/go-metrics v0.3.8/go.mod h1:4O98XIr/9W0sxpJ8UaYkvjk10Iff7SnFrb4QAOwNTFc= github.com/armon/go-metrics v0.4.1 h1:hR91U9KYmb6bLBYLQjyM+3j+rcd/UhE+G78SFnF8gJA= github.com/armon/go-metrics v0.4.1/go.mod h1:E6amYzXo6aW1tqzoZGT755KkbgrJsSdpwZ+3JqfkOG4= github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4= +github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= @@ -33,14 +38,13 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dgate-io/chi-router v0.0.0-20231217131951-d154152d5115 h1:AVEnGd1UBqJU7MnbyAtPfp47mlI5GvMS4fFNZVMS0KA= github.com/dgate-io/chi-router v0.0.0-20231217131951-d154152d5115/go.mod h1:MyLj6L03q1t8GW/541pHuP6co58QfLppSYPS0PvLtC8= -github.com/dgate-io/raft-badger v0.0.0-20231217131807-c5eb3f9eafa5 h1:clmNs28JV+F8rV6cXhByKOvQIMWLNfIvOrBD9nxkUUE= -github.com/dgate-io/raft-badger v0.0.0-20231217131807-c5eb3f9eafa5/go.mod h1:vXdEq7albbhewiVglqyJ4+gTlGhRVVYKO4jhKv0qVuk= github.com/dgraph-io/badger/v4 v4.2.0 h1:kJrlajbXXL9DFTNuhhu9yCx7JJa4qpYWxtE8BzuWsEs= github.com/dgraph-io/badger/v4 v4.2.0/go.mod h1:qfCqhPoWDFJRx1gp5QwwyGo8xk1lbHUxvK9nK0OGAak= github.com/dgraph-io/ristretto v0.1.1 h1:6CWw5tJNgpegArSHpNHJKldNeq03FQCwYvfMVWajOK8= github.com/dgraph-io/ristretto v0.1.1/go.mod h1:S1GPSBCYCIhmVNfcth17y2zZtQT6wzkzgwUve0VDWWA= -github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA= github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw= +github.com/dgryski/go-farm v0.0.0-20191112170834-c2139c5d712b h1:SeiGBzKrEtuDddnBABHkp4kq9sBGE9nuYmk6FPTg0zg= +github.com/dgryski/go-farm v0.0.0-20191112170834-c2139c5d712b/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw= github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc= github.com/dlclark/regexp2 v1.7.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= github.com/dlclark/regexp2 v1.10.0 h1:+/GIL799phkJqYW+3YbOd8LCcbHzT0Pbo8zl70MHsq0= @@ -61,8 +65,8 @@ github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.m github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98= github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk= -github.com/fatih/color v1.14.1 h1:qfhVLaG5s+nCROl1zJsZRxFeYrHLqWroPOQ8BWiNb4w= -github.com/fatih/color v1.14.1/go.mod h1:2oHN61fhTpgcxD3TSWCgKDiH1+x4OiDVVGH8WlgGZGg= +github.com/fatih/color v1.17.0 h1:GlRw1BRJxkpqUCBKzKOw098ed57fEsKeNjpTe3cSjK4= +github.com/fatih/color v1.17.0/go.mod h1:YZ7TlrGPkiz6ku9fK3TLD/pl3CpsiFyu8N92HLgmosI= github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA= github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM= github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= @@ -121,21 +125,30 @@ github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+ github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4= github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/hashicorp/go-cleanhttp v0.5.0/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80= -github.com/hashicorp/go-hclog v1.6.2 h1:NOtoftovWkDheyUM/8JW3QMiXyxJK3uHRK7wV04nD2I= -github.com/hashicorp/go-hclog v1.6.2/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= +github.com/hashicorp/go-hclog v0.9.1/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ= +github.com/hashicorp/go-hclog v1.6.3 h1:Qr2kF+eVWjTiYmU7Y31tYlP1h0q/X3Nl3tPGdaB11/k= +github.com/hashicorp/go-hclog v1.6.3/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60= github.com/hashicorp/go-immutable-radix v1.3.1 h1:DKHmCUm2hRBK510BaiZlwvpD40f8bJFeZnpfm2KLowc= github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60= -github.com/hashicorp/go-msgpack/v2 v2.1.1 h1:xQEY9yB2wnHitoSzk/B9UjXWRQ67QKu5AOm8aFp8N3I= -github.com/hashicorp/go-msgpack/v2 v2.1.1/go.mod h1:upybraOAblm4S7rx0+jeNy+CWWhzywQsSRV5033mMu4= +github.com/hashicorp/go-msgpack v0.5.5/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM= +github.com/hashicorp/go-msgpack v1.1.5 h1:9byZdVjKTe5mce63pRVNP1L7UAmdHOTEMGehn6KvJWs= +github.com/hashicorp/go-msgpack v1.1.5/go.mod h1:gWVc3sv/wbDmR3rQsj1CAktEZzoz1YNK9NfGLXJ69/4= +github.com/hashicorp/go-msgpack/v2 v2.1.2 h1:4Ee8FTp834e+ewB71RDrQ0VKpyFdrKOjvYtnQ/ltVj0= +github.com/hashicorp/go-msgpack/v2 v2.1.2/go.mod h1:upybraOAblm4S7rx0+jeNy+CWWhzywQsSRV5033mMu4= github.com/hashicorp/go-retryablehttp v0.5.3/go.mod h1:9B5zBasrRhHXnJnui7y6sL7es7NDiJgTc6Er0maI1Xs= github.com/hashicorp/go-uuid v1.0.0 h1:RS8zrF7PhGwyNPOtxSClXXj9HA8feRnJzgnI1RJCSnM= github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= -github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc= -github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= -github.com/hashicorp/raft v1.6.0 h1:tkIAORZy2GbJ2Trp5eUSggLXDPOJLXC+JJLNMMqtgtM= -github.com/hashicorp/raft v1.6.0/go.mod h1:Xil5pDgeGwRWuX4uPUmwa+7Vagg4N804dz6mhNi6S7o= +github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= +github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= +github.com/hashicorp/raft v1.1.0/go.mod h1:4Ak7FSPnuvmb0GV6vgIAJ4vYT4bek9bb6Q+7HVbyzqM= +github.com/hashicorp/raft v1.7.0 h1:4u24Qn6lQ6uwziM++UgsyiT64Q8GyRn43CV41qPiz1o= +github.com/hashicorp/raft v1.7.0/go.mod h1:N1sKh6Vn47mrWvEArQgILTyng8GoDRNYlgKyK7PMjs0= +github.com/hashicorp/raft-boltdb v0.0.0-20231211162105-6c830fa4535e h1:SK4y8oR4ZMHPvwVHryKI88kJPJda4UyWYvG5A6iEQxc= +github.com/hashicorp/raft-boltdb v0.0.0-20231211162105-6c830fa4535e/go.mod h1:EMz/UIuG93P0MBeHh6CbXQAEe8ckVJLZjhD17lBzK5Q= +github.com/hashicorp/raft-boltdb/v2 v2.3.0 h1:fPpQR1iGEVYjZ2OELvUHX600VAK5qmdnDEv3eXOwZUA= +github.com/hashicorp/raft-boltdb/v2 v2.3.0/go.mod h1:YHukhB04ChJsLHLJEUD6vjFyLX2L3dsX3wPBZcX4tmc= github.com/ianlancetaylor/demangle v0.0.0-20220319035150-800ac71e25c2/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w= github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU= github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= @@ -178,8 +191,8 @@ github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovk github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU= github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94= github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= -github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA= -github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw= github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s= @@ -204,6 +217,7 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= +github.com/prometheus/client_golang v0.9.2/go.mod h1:OsXs2jCmiKlQ1lTBmv21f2mNfw4xf/QclQDMrYNZzcM= github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo= github.com/prometheus/client_golang v1.4.0/go.mod h1:e9GMxYsXl05ICDXkRhurwBS4Q3OK1iX/F2sw+iXX5zU= github.com/prometheus/client_golang v1.19.0 h1:ygXvpU1AoN1MhdzckN+PyD9QJOSD4x7kmXYlnfbA6JU= @@ -214,11 +228,13 @@ github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1: github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E= github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY= +github.com/prometheus/common v0.0.0-20181126121408-4724e9255275/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro= github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4= github.com/prometheus/common v0.9.1/go.mod h1:yhUN8i9wzaXS3w1O07YhxHEBxD+W35wd8bs7vj7HSQ4= github.com/prometheus/common v0.48.0 h1:QO8U2CdOzSn1BBsmXJXduaaW+dY/5QLjfB8svtSzKKE= github.com/prometheus/common v0.48.0/go.mod h1:0/KsvlIEfPQCQ5I2iNSAWKPZziNCvRs5EC6ILDTlAPc= github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= +github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA= github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A= github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo= @@ -259,6 +275,8 @@ github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673/go.mod h1:N3UwUGtsr github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +go.etcd.io/bbolt v1.3.10 h1:+BqfJTcCzTItrop8mq/lbzL8wSGtj94UO/3U31shqG0= +go.etcd.io/bbolt v1.3.10/go.mod h1:bK3UQLPJZly7IlNmV7uVHJDxfe5aK9Ll93e/74Y9oEQ= go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0= go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo= go.opentelemetry.io/otel v1.26.0 h1:LQwgL5s/1W7YiiRwxf03QGnWLb2HW4pLiAhaA5cZXBs= @@ -294,6 +312,7 @@ golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= @@ -314,6 +333,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M= +golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -335,8 +356,8 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20221010170243-090e33056c14/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o= -golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.19.0 h1:+ThwsDv+tYfnJFhF4L8jITxu1tdTWRTZpdsWgEgjL6Q= @@ -352,6 +373,7 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY= golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= +golang.org/x/tools v0.0.0-20190424220101-1e8e1cfdf96b/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= diff --git a/internal/admin/admin_fsm.go b/internal/admin/admin_fsm.go index 781cd17..d2612a8 100644 --- a/internal/admin/admin_fsm.go +++ b/internal/admin/admin_fsm.go @@ -2,6 +2,7 @@ package admin import ( "encoding/json" + "errors" "io" "github.com/dgate-io/dgate/internal/admin/changestate" @@ -13,43 +14,39 @@ import ( type dgateAdminFSM struct { cs changestate.ChangeState logger *zap.Logger - index uint64 } var _ raft.BatchingFSM = (*dgateAdminFSM)(nil) func newDGateAdminFSM(logger *zap.Logger, cs changestate.ChangeState) *dgateAdminFSM { - return &dgateAdminFSM{cs, logger, 0} + return &dgateAdminFSM{cs, logger} } -func (fsm *dgateAdminFSM) SetIndex(index uint64) { - fsm.index = index -} - -func (fsm *dgateAdminFSM) applyLog(log *raft.Log, replay bool) (*spec.ChangeLog, error) { - log.Index = fsm.index +func (fsm *dgateAdminFSM) applyLog(log *raft.Log, reload bool) (*spec.ChangeLog, error) { switch log.Type { case raft.LogCommand: var cl spec.ChangeLog if err := json.Unmarshal(log.Data, &cl); err != nil { fsm.logger.Error("Error unmarshalling change log", zap.Error(err)) return nil, err + } + + if cl.ID == "" { + fsm.logger.Error("Change log ID is empty") + return nil, errors.New("change log ID is empty") } else if cl.Cmd.IsNoop() { return nil, nil - } else if cl.ID == "" { - fsm.logger.Error("Change log ID is empty", zap.Error(err)) - panic("change log ID is empty") } // find a way to only reload if latest index to save time - return &cl, fsm.cs.ProcessChangeLog(&cl, replay) + return &cl, fsm.cs.ProcessChangeLog(&cl, reload) case raft.LogConfiguration: servers := raft.DecodeConfiguration(log.Data).Servers - for i, server := range servers { - fsm.logger.Debug("configuration update server", - zap.Any("address", server.Address), - zap.Int("index", i), - ) - } + fsm.logger.Debug("configuration update server", + zap.Any("address", servers), + zap.Uint64("index", log.Index), + zap.Uint64("term", log.Term), + zap.Time("appended", log.AppendedAt), + ) default: fsm.logger.Error("Unknown log type in FSM Apply") } @@ -57,6 +54,13 @@ func (fsm *dgateAdminFSM) applyLog(log *raft.Log, replay bool) (*spec.ChangeLog, } func (fsm *dgateAdminFSM) Apply(log *raft.Log) any { + rft := fsm.cs.Raft() + fsm.logger.Debug("apply single log", + zap.Uint64("applied", rft.AppliedIndex()), + zap.Uint64("commit", rft.CommitIndex()), + zap.Uint64("last", rft.LastIndex()), + zap.Uint64("logIndex", log.Index), + ) _, err := fsm.applyLog(log, true) return err } @@ -68,7 +72,6 @@ func (fsm *dgateAdminFSM) ApplyBatch(logs []*raft.Log) []any { zap.Uint64("applied", rft.AppliedIndex()), zap.Uint64("commit", rft.CommitIndex()), zap.Uint64("last", rft.LastIndex()), - zap.Uint64("fsmLastIndex", fsm.index), zap.Uint64("log[0]", logs[0].Index), zap.Uint64("log[-1]", logs[lastIndex].Index), zap.Int("logs", len(logs)), @@ -76,17 +79,22 @@ func (fsm *dgateAdminFSM) ApplyBatch(logs []*raft.Log) []any { results := make([]any, len(logs)) for i, log := range logs { // TODO: check to see if this can be optimized channels raft node provides - _, results[i] = fsm.applyLog( - log, lastIndex == i, - ) + _, err := fsm.applyLog(log, lastIndex == i) + if err != nil { + fsm.logger.Error("Error applying log", zap.Error(err)) + results[i] = err + } } return results } func (fsm *dgateAdminFSM) Snapshot() (raft.FSMSnapshot, error) { - panic("snapshots not supported") + fsm.cs = nil + fsm.logger.Warn("snapshots not supported") + return nil, errors.New("snapshots not supported") } func (fsm *dgateAdminFSM) Restore(rc io.ReadCloser) error { - panic("snapshots not supported") + fsm.logger.Warn("snapshots not supported, cannot restore") + return nil } diff --git a/internal/admin/admin_raft.go b/internal/admin/admin_raft.go index f9d126f..cdaa743 100644 --- a/internal/admin/admin_raft.go +++ b/internal/admin/admin_raft.go @@ -3,6 +3,7 @@ package admin import ( "context" "fmt" + "math" "net" "net/http" "path" @@ -13,12 +14,11 @@ import ( "github.com/dgate-io/dgate/internal/config" "github.com/dgate-io/dgate/pkg/raftadmin" "github.com/dgate-io/dgate/pkg/rafthttp" - "github.com/dgate-io/dgate/pkg/spec" "github.com/dgate-io/dgate/pkg/storage" + "github.com/dgate-io/dgate/pkg/util" "github.com/dgate-io/dgate/pkg/util/logadapter" - raftbadgerdb "github.com/dgate-io/raft-badger" - "github.com/dgraph-io/badger/v4" "github.com/hashicorp/raft" + boltdb "github.com/hashicorp/raft-boltdb/v2" "go.uber.org/zap" ) @@ -29,31 +29,44 @@ func setupRaft( cs changestate.ChangeState, ) { adminConfig := conf.AdminConfig - var sstore raft.StableStore - var lstore raft.LogStore + var logStore raft.LogStore + var configStore raft.StableStore + var snapStore raft.SnapshotStore switch conf.Storage.StorageType { case config.StorageTypeMemory: - sstore = raft.NewInmemStore() - lstore = raft.NewInmemStore() + logStore = raft.NewInmemStore() + configStore = raft.NewInmemStore() case config.StorageTypeFile: fileConfig, err := config.StoreConfig[storage.FileStoreConfig](conf.Storage.Config) if err != nil { panic(fmt.Errorf("invalid config: %s", err)) } - badgerLogger := logadapter.NewZap2BadgerAdapter(logger.Named("badger-file")) - raftDir := path.Join(fileConfig.Directory, "raft") - badgerStore, err := raftbadgerdb.New( - badger.DefaultOptions(raftDir). - WithLogger(badgerLogger), + raftDir := path.Join(fileConfig.Directory) + + snapStore, err = raft.NewFileSnapshotStore( + path.Join(raftDir), 5, + zap.NewStdLog(logger.Named("snap-file")).Writer(), ) if err != nil { panic(err) } - sstore = badgerStore - lstore = badgerStore + if boltStore, err := boltdb.NewBoltStore( + path.Join(raftDir, "raft.db"), + ); err != nil { + panic(err) + } else { + configStore = boltStore + logStore = boltStore + } default: panic(fmt.Errorf("invalid storage type: %s", conf.Storage.StorageType)) } + + logger.Info("raft store", + zap.Stringer("storage_type", conf.Storage.StorageType), + zap.Any("storage_config", conf.Storage.Config), + ) + raftConfig := adminConfig.Replication.LoadRaftConfig( &raft.Config{ ProtocolVersion: raft.ProtocolVersionMax, @@ -62,11 +75,12 @@ func setupRaft( ElectionTimeout: time.Second * 5, CommitTimeout: time.Second * 4, BatchApplyCh: false, - MaxAppendEntries: 512, + MaxAppendEntries: 1024, LeaderLeaseTimeout: time.Second * 4, + // TODO: Support snapshots - SnapshotInterval: time.Hour*2 ^ 32, - SnapshotThreshold: ^uint64(0), + SnapshotInterval: time.Duration(9999 * time.Hour), + SnapshotThreshold: math.MaxUint64, Logger: logadapter.NewZap2HCLogAdapter(logger), }, ) @@ -87,19 +101,16 @@ func setupRaft( adminConfig.Replication.AdvertScheme+"://(address)/raft", ) fsmLogger := logger.Named("fsm") - snapstore := raft.NewInmemSnapshotStore() - fsm := newDGateAdminFSM(fsmLogger, cs) + adminFSM := newDGateAdminFSM(fsmLogger, cs) raftNode, err := raft.NewRaft( - raftConfig, fsm, lstore, - sstore, snapstore, transport, + raftConfig, adminFSM, logStore, + configStore, snapStore, transport, ) if err != nil { panic(err) } - observerChan := make(chan raft.Observation, 10) - raftNode.RegisterObserver(raft.NewObserver(observerChan, false, nil)) - cs.SetupRaft(raftNode, observerChan) + cs.SetupRaft(raftNode) // Setup raft handler server.Handle("/raft/*", transport) @@ -109,7 +120,7 @@ func setupRaft( raftNode, raftAdminLogger, []raft.ServerAddress{address}, ) - // Setup handler raft + // Setup handler for raft admin server.HandleFunc("/raftadmin/*", func(w http.ResponseWriter, r *http.Request) { if adminConfig.Replication.SharedKey != "" { sharedKey := r.Header.Get("X-DGate-Shared-Key") @@ -121,6 +132,12 @@ func setupRaft( raftAdmin.ServeHTTP(w, r) }) + // Setup handler for stats + server.Handle("/raftadmin/stats", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Raft-State", raftNode.State().String()) + util.JsonResponse(w, http.StatusOK, raftNode.Stats()) + })) + configFuture := raftNode.GetConfiguration() if err = configFuture.Error(); err != nil { panic(err) @@ -136,8 +153,6 @@ func setupRaft( zap.Int("config_proto", int(raftConfig.ProtocolVersion)), ) - defer cs.ProcessChangeLog(spec.NewNoopChangeLog(), false) - if adminConfig.Replication.BootstrapCluster && len(serverConfig.Servers) == 0 { logger.Info("bootstrapping cluster", zap.String("id", raftId), diff --git a/internal/admin/changestate/change_state.go b/internal/admin/changestate/change_state.go index 882288d..e3ad809 100644 --- a/internal/admin/changestate/change_state.go +++ b/internal/admin/changestate/change_state.go @@ -10,17 +10,17 @@ import ( type ChangeState interface { // Change state ApplyChangeLog(cl *spec.ChangeLog) error - ProcessChangeLog(*spec.ChangeLog, bool) error + ProcessChangeLog(cl *spec.ChangeLog, reload bool) error WaitForChanges() error ReloadState(bool, ...*spec.ChangeLog) error - ChangeHash() uint32 + ChangeHash() uint64 ChangeLogs() []*spec.ChangeLog // Readiness Ready() bool // Replication - SetupRaft(*raft.Raft, chan raft.Observation) + SetupRaft(*raft.Raft) Raft() *raft.Raft // Resources diff --git a/internal/admin/changestate/testutil/change_state.go b/internal/admin/changestate/testutil/change_state.go index d199958..62205c0 100644 --- a/internal/admin/changestate/testutil/change_state.go +++ b/internal/admin/changestate/testutil/change_state.go @@ -22,8 +22,8 @@ func (m *MockChangeState) ApplyChangeLog(cl *spec.ChangeLog) error { } // ChangeHash implements changestate.ChangeState. -func (m *MockChangeState) ChangeHash() uint32 { - return m.Called().Get(0).(uint32) +func (m *MockChangeState) ChangeHash() uint64 { + return m.Called().Get(0).(uint64) } // DocumentManager implements changestate.ChangeState. @@ -66,7 +66,7 @@ func (m *MockChangeState) ReloadState(a bool, cls ...*spec.ChangeLog) error { } // SetupRaft implements changestate.ChangeState. -func (m *MockChangeState) SetupRaft(*raft.Raft, chan raft.Observation) { +func (m *MockChangeState) SetupRaft(*raft.Raft) { m.Called().Error(0) } diff --git a/internal/admin/routes/misc_routes.go b/internal/admin/routes/misc_routes.go index 636ffaa..cab2de1 100644 --- a/internal/admin/routes/misc_routes.go +++ b/internal/admin/routes/misc_routes.go @@ -3,6 +3,7 @@ package routes import ( "encoding/json" "net/http" + "strconv" "github.com/dgate-io/chi-router" "github.com/dgate-io/dgate/internal/admin/changestate" @@ -11,35 +12,23 @@ import ( ) func ConfigureChangeLogAPI(server chi.Router, cs changestate.ChangeState, appConfig *config.DGateConfig) { - server.Get("/changelog/hash", func(w http.ResponseWriter, r *http.Request) { - if err := cs.WaitForChanges(); err != nil { - util.JsonError(w, http.StatusInternalServerError, err.Error()) - return + server.Get("/changelog", func(w http.ResponseWriter, r *http.Request) { + hash := cs.ChangeHash() + logs := cs.ChangeLogs() + lastLogId := "" + if len(logs) > 0 { + lastLogId = logs[len(logs)-1].ID } - - if b, err := json.Marshal(map[string]any{ - "hash": cs.ChangeHash(), - }); err != nil { + b, err := json.Marshal(map[string]any{ + "count": len(logs), + "hash": strconv.FormatUint(hash, 36), + "latest": lastLogId, + }) + if err != nil { util.JsonError(w, http.StatusInternalServerError, err.Error()) - } else { - w.Header().Set("Content-Type", "application/json") - w.Write([]byte(b)) - } - }) - server.Get("/changelog/count", func(w http.ResponseWriter, r *http.Request) { - if err := cs.WaitForChanges(); err != nil { - util.JsonError(w, http.StatusInternalServerError, err.Error()) - return - } - - if b, err := json.Marshal(map[string]any{ - "count": len(cs.ChangeLogs()), - }); err != nil { - util.JsonError(w, http.StatusInternalServerError, err.Error()) - } else { - w.Header().Set("Content-Type", "application/json") - w.Write([]byte(b)) } + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(b)) }) } @@ -54,18 +43,26 @@ func ConfigureHealthAPI(server chi.Router, version string, cs changestate.Change server.Get("/readyz", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") - if r := cs.Raft(); r != nil { - w.Header().Set("X-Raft-State", r.State().String()) - if r.Leader() == "" { - w.WriteHeader(http.StatusServiceUnavailable) - w.Write([]byte(`{"status":"no leader"}`)) - return - } else if !cs.Ready() { - w.WriteHeader(http.StatusServiceUnavailable) - w.Write([]byte(`{"status":"not ready"}`)) - return + if cs.Ready() { + if r := cs.Raft(); r != nil { + if err := cs.WaitForChanges(); err != nil { + w.WriteHeader(http.StatusServiceUnavailable) + w.Write([]byte(`{"status":"not ready"}`)) + return + } + w.Header().Set("X-Raft-State", r.State().String()) + if leaderAddr := r.Leader(); leaderAddr == "" { + w.WriteHeader(http.StatusServiceUnavailable) + w.Write([]byte(`{"status":"no leader"}`)) + return + } else { + w.Header().Set("X-Raft-Leader", string(leaderAddr)) + } } + w.Write(healthlyResp) + } else { + w.WriteHeader(http.StatusServiceUnavailable) + w.Write([]byte(`{"status":"not ready"}`)) } - w.Write(healthlyResp) }) } diff --git a/internal/config/config.go b/internal/config/config.go index cf1c0c5..b95e427 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -44,7 +44,7 @@ type ( TLS *DGateTLSConfig `koanf:"tls"` EnableH2C bool `koanf:"enable_h2c"` EnableHTTP2 bool `koanf:"enable_http2"` - EnableConsoleLogger bool `koanf:"enable_console_logger"` + ConsoleLogLevel string `koanf:"console_log_level"` RedirectHttpsDomains []string `koanf:"redirect_https"` AllowedDomains []string `koanf:"allowed_domains"` GlobalHeaders map[string]string `koanf:"global_headers"` @@ -234,6 +234,7 @@ func (conf *DGateConfig) GetLogger() (*zap.Logger, error) { config.Development = conf.Debug config.EncoderConfig.EncodeTime = zapcore.RFC3339TimeEncoder config.OutputPaths = []string{"stdout"} + config.Sampling = nil if config.EncoderConfig.EncodeLevel = zapcore.CapitalLevelEncoder; conf.LogColor { config.EncoderConfig.EncodeLevel = zapcore.CapitalColorLevelEncoder diff --git a/internal/config/configtest/dgate_configs.go b/internal/config/configtest/dgate_configs.go index 36956f3..82e8de9 100644 --- a/internal/config/configtest/dgate_configs.go +++ b/internal/config/configtest/dgate_configs.go @@ -18,7 +18,7 @@ func NewTestDGateConfig() *config.DGateConfig { Version: "v1", Tags: []string{"test"}, Storage: config.DGateStorageConfig{ - StorageType: config.StorageTypeDebug, + StorageType: config.StorageTypeMemory, }, ProxyConfig: config.DGateProxyConfig{ AllowedDomains: []string{"*test.com", "localhost"}, diff --git a/internal/config/loader.go b/internal/config/loader.go index 533f22c..9a2ab9a 100644 --- a/internal/config/loader.go +++ b/internal/config/loader.go @@ -156,6 +156,7 @@ func LoadConfig(dgateConfigPath string) (*DGateConfig, error) { kDefault(k, "proxy.port", 80) kDefault(k, "proxy.enable_h2c", false) kDefault(k, "proxy.enable_http2", false) + kDefault(k, "proxy.console_log_level", k.Get("log_level")) if k.Get("proxy.enable_h2c") == true && k.Get("proxy.enable_http2") == false { diff --git a/internal/config/store_config.go b/internal/config/store_config.go index 808170c..0b1705e 100644 --- a/internal/config/store_config.go +++ b/internal/config/store_config.go @@ -7,7 +7,6 @@ import ( type StorageType string const ( - StorageTypeDebug StorageType = "debug" StorageTypeMemory StorageType = "memory" StorageTypeFile StorageType = "file" ) @@ -26,3 +25,7 @@ func StoreConfig[T any, C any](config C) (T, error) { err = decoder.Decode(config) return output, err } + +func (st StorageType) String() string { + return string(st) +} diff --git a/internal/proxy/change_log.go b/internal/proxy/change_log.go index 1fab512..ccd3f49 100644 --- a/internal/proxy/change_log.go +++ b/internal/proxy/change_log.go @@ -2,125 +2,103 @@ package proxy import ( "fmt" - "strconv" "time" "errors" "github.com/dgate-io/dgate/pkg/spec" "github.com/dgate-io/dgate/pkg/util/sliceutil" - "github.com/dgraph-io/badger/v4" + "github.com/hashicorp/raft" "github.com/mitchellh/mapstructure" "go.uber.org/zap" ) // processChangeLog - processes a change log and applies the change to the proxy state func (ps *ProxyState) processChangeLog(cl *spec.ChangeLog, reload, store bool) (err error) { - defer func() { - if err != nil && !cl.Cmd.IsNoop() { - ps.ready.Store(true) - if changeHash, err := HashAny(ps.changeHash, cl); err != nil { - ps.logger.Error("error hashing change log", zap.Error(err)) - return - } else { - ps.changeHash = changeHash + if reload { + defer func(start time.Time) { + ps.logger.Debug("processing change log", + zap.String("id", cl.ID), + zap.Duration("duration", time.Since(start)), + ) + }(time.Now()) + } + ps.proxyLock.Lock() + defer ps.proxyLock.Unlock() + + // store change log if there is no error + if store && !cl.Cmd.IsNoop() { + defer func() { + if err == nil { + if !ps.replicationEnabled { + // dont store change logs + if err = ps.store.StoreChangeLog(cl); err != nil { + ps.logger.Error("Error storing change log, restarting state", zap.Error(err)) + return + } + } + // in memory storage for state restarts + ps.changeLogs = append(ps.changeLogs, cl) } - } - }() - if !cl.Cmd.IsNoop() { - if len(ps.changeLogs) > 0 { - xcl := ps.changeLogs[len(ps.changeLogs)-1] - if xcl.ID == cl.ID { - ps.logger.Debug("duplicate change log", zap.String("id", cl.ID)) + }() + } + + if len(ps.changeLogs) > 0 { + xcl := ps.changeLogs[len(ps.changeLogs)-1] + if xcl.ID == cl.ID { + if r := ps.Raft(); r != nil && r.State() == raft.Leader { + // FYI: we still need to store the change log return nil } + ps.logger.Error("duplicate change log", + zap.String("id", cl.ID), + zap.Stringer("cmd", cl.Cmd), + ) + return errors.New("duplicate change log") } - strconv.FormatInt(time.Now().UnixNano(), 36) - ps.changeLogs = append(ps.changeLogs, cl) - switch cl.Cmd.Resource() { - case spec.Namespaces: - var item spec.Namespace - item, err = decode[spec.Namespace](cl.Item) - if err == nil { - ps.logger.Debug("Processing namespace", zap.String("name", item.Name)) - err = ps.processNamespace(&item, cl) - } - case spec.Services: - var item spec.Service - item, err = decode[spec.Service](cl.Item) - if err == nil { - ps.logger.Debug("Processing service", zap.String("name", item.Name)) - err = ps.processService(&item, cl) - } - case spec.Routes: - var item spec.Route - item, err = decode[spec.Route](cl.Item) - if err == nil { - ps.logger.Debug("Processing route", zap.String("name", item.Name)) - err = ps.processRoute(&item, cl) - } - case spec.Modules: - var item spec.Module - item, err = decode[spec.Module](cl.Item) - if err == nil { - ps.logger.Debug("Processing module", zap.String("name", item.Name)) - err = ps.processModule(&item, cl) - } - case spec.Domains: - var item spec.Domain - item, err = decode[spec.Domain](cl.Item) - if err == nil { - ps.logger.Debug("Processing domain", zap.String("name", item.Name)) - err = ps.processDomain(&item, cl) - } - case spec.Collections: - var item spec.Collection - item, err = decode[spec.Collection](cl.Item) - if err == nil { - ps.logger.Debug("Processing collection", zap.String("name", item.Name)) - err = ps.processCollection(&item, cl) - } - case spec.Documents: - var item spec.Document - item, err = decode[spec.Document](cl.Item) - if err == nil { - ps.logger.Debug("Processing document", zap.String("id", item.ID)) - err = ps.processDocument(&item, cl) - } - case spec.Secrets: - var item spec.Secret - item, err = decode[spec.Secret](cl.Item) + } + + // apply change log to the state + if !cl.Cmd.IsNoop() { + defer func() { if err == nil { - ps.logger.Debug("Processing secret", zap.String("name", item.Name)) - err = ps.processSecret(&item, cl) + hash_retry: + oldHash := ps.changeHash.Load() + if newHash, err := HashAny(oldHash, cl.ID); err != nil { + ps.logger.Error("error hashing change log", zap.Error(err)) + } else if !ps.changeHash.CompareAndSwap(oldHash, newHash) { + goto hash_retry + } + } else { + ps.restartState(func(err error) { + if err != nil { + go ps.Stop() + } + }) } - default: - err = fmt.Errorf("unknown command: %s", cl.Cmd) - } - if err != nil { + }() + if cl.Cmd.Resource() == spec.Documents && !store { + return nil + } else if err = ps.processResource(cl); err != nil { ps.logger.Error("decoding or processing change log", zap.Error(err)) return } } + + // apply state changes to the proxy if reload { - if cl.Cmd.IsNoop() || cl.Cmd.Resource().IsRelatedTo(spec.Routes) { - ps.logger.Debug("Registering change log", zap.Stringer("cmd", cl.Cmd)) - if err = ps.reconfigureState(false); err != nil { + overrideReload := cl.Cmd.IsNoop() || ps.pendingChanges + if overrideReload || cl.Cmd.Resource().IsRelatedTo(spec.Routes) { + ps.logger.Debug("Reloading change log at...", zap.String("id", cl.ID)) + if err = ps.reconfigureState(cl); err != nil { ps.logger.Error("Error registering change log", zap.Error(err)) return } + ps.ready.CompareAndSwap(false, true) + ps.pendingChanges = false } - } - if store { - if err = ps.store.StoreChangeLog(cl); err != nil { - ps.logger.Error("Error storing change log, restarting state", zap.Error(err)) - ps.restartState(func(err error) { - if err != nil { - go ps.Stop() - } - }) - return - } + } else if !cl.Cmd.IsNoop() { + ps.pendingChanges = true } return nil @@ -145,6 +123,54 @@ func decode[T any](input any) (T, error) { return output, nil } +func (ps *ProxyState) processResource(cl *spec.ChangeLog) (err error) { + switch cl.Cmd.Resource() { + case spec.Namespaces: + var item spec.Namespace + if item, err = decode[spec.Namespace](cl.Item); err == nil { + err = ps.processNamespace(&item, cl) + } + case spec.Services: + var item spec.Service + if item, err = decode[spec.Service](cl.Item); err == nil { + err = ps.processService(&item, cl) + } + case spec.Routes: + var item spec.Route + if item, err = decode[spec.Route](cl.Item); err == nil { + err = ps.processRoute(&item, cl) + } + case spec.Modules: + var item spec.Module + if item, err = decode[spec.Module](cl.Item); err == nil { + err = ps.processModule(&item, cl) + } + case spec.Domains: + var item spec.Domain + if item, err = decode[spec.Domain](cl.Item); err == nil { + err = ps.processDomain(&item, cl) + } + case spec.Collections: + var item spec.Collection + if item, err = decode[spec.Collection](cl.Item); err == nil { + err = ps.processCollection(&item, cl) + } + case spec.Documents: + var item spec.Document + if item, err = decode[spec.Document](cl.Item); err == nil { + err = ps.processDocument(&item, cl) + } + case spec.Secrets: + var item spec.Secret + if item, err = decode[spec.Secret](cl.Item); err == nil { + err = ps.processSecret(&item, cl) + } + default: + err = fmt.Errorf("unknown command: %s", cl.Cmd) + } + return err +} + func (ps *ProxyState) processNamespace(ns *spec.Namespace, cl *spec.ChangeLog) error { switch cl.Cmd.Action() { case spec.Add: @@ -262,40 +288,30 @@ func (ps *ProxyState) processSecret(scrt *spec.Secret, cl *spec.ChangeLog) (err return err } +// restoreFromChangeLogs - restores the proxy state from change logs; directApply is used to avoid locking the proxy state func (ps *ProxyState) restoreFromChangeLogs(directApply bool) error { - logs, err := ps.store.FetchChangeLogs() - if err != nil { - if err == badger.ErrKeyNotFound { - ps.logger.Debug("no state change logs found in storage") - } else { - return errors.New("failed to get state change logs from storage: " + err.Error()) - } + if logs, err := ps.store.FetchChangeLogs(); err != nil { + return errors.New("failed to get state change logs from storage: " + err.Error()) } else { ps.logger.Info("restoring state change logs from storage", zap.Int("count", len(logs))) // we might need to sort the change logs by timestamp - for i, cl := range logs { - ps.logger.Debug("restoring change log", - zap.Int("index", i), - zap.Stringer("changeLog", cl.Cmd), - ) - err = ps.processChangeLog(cl, false, false) - if err != nil { - if ps.config.Debug { - ps.logger.Error("error restorng from change logs", zap.Error(err)) - continue - } + for _, cl := range logs { + // ps.logger.Debug("restoring change log", + // zap.Int("index", i), + // zap.Stringer("changeLog", cl.Cmd), + // ) + if err = ps.processChangeLog(cl, false, false); err != nil { return err + } else { + ps.changeLogs = append(ps.changeLogs, cl) } } - if !directApply { - cl := spec.NewNoopChangeLog() - if err = ps.processChangeLog(cl, true, false); err != nil { + if cl := spec.NewNoopChangeLog(); !directApply { + if err = ps.reconfigureState(cl); err != nil { return err } - } else { - if err = ps.reconfigureState(false); err != nil { - return nil - } + } else if err = ps.processChangeLog(cl, true, false); err != nil { + return err } // TODO: optionally compact change logs through a flag in config? diff --git a/internal/proxy/dynamic_proxy.go b/internal/proxy/dynamic_proxy.go index d635155..fb94edd 100644 --- a/internal/proxy/dynamic_proxy.go +++ b/internal/proxy/dynamic_proxy.go @@ -1,6 +1,7 @@ package proxy import ( + "context" "errors" "fmt" "net/http" @@ -15,9 +16,10 @@ import ( "go.uber.org/zap" "golang.org/x/net/http2" "golang.org/x/net/http2/h2c" + "golang.org/x/sync/errgroup" ) -func (ps *ProxyState) reconfigureState(init bool) (err error) { +func (ps *ProxyState) reconfigureState(log *spec.ChangeLog) (err error) { defer func() { if err != nil { ps.restartState(func(err error) { @@ -29,80 +31,91 @@ func (ps *ProxyState) reconfigureState(init bool) (err error) { } }() - ps.proxyLock.Lock() - defer ps.proxyLock.Unlock() - start := time.Now() - if err = ps.setupModules(); err != nil { + if err = ps.setupModules(log); err != nil { + ps.logger.Error("Error setting up modules", zap.Error(err)) return } - if err = ps.setupRoutes(); err != nil { + if err = ps.setupRoutes(log); err != nil { + ps.logger.Error("Error setting up routes", zap.Error(err)) return } elapsed := time.Since(start) - if !init { - ps.logger.Debug("State reloaded", - zap.Duration("elapsed", elapsed), - ) - } else { - ps.logger.Info("State initialized", - zap.Duration("elapsed", elapsed), - ) - } + ps.logger.Debug("State reloaded", + zap.Duration("elapsed", elapsed), + ) return nil } -func (ps *ProxyState) setupModules() error { - ps.logger.Debug("Setting up modules") - for _, route := range ps.rm.GetRoutes() { - if len(route.Modules) > 0 { - mod := route.Modules[0] - var ( - err error - program *goja.Program - modPayload string = mod.Payload - ) - start := time.Now() - if mod.Type == spec.ModuleTypeTypescript { - if modPayload, err = typescript.Transpile(modPayload); err != nil { - ps.logger.Error("Error transpiling module: " + mod.Name) - return err +func (ps *ProxyState) setupModules(log *spec.ChangeLog) error { + var routes = []*spec.DGateRoute{} + if log.Namespace == "" || ps.pendingChanges { + routes = ps.rm.GetRoutes() + } else { + routes = ps.rm.GetRoutesByNamespace(log.Namespace) + } + programMap := make(map[string]*goja.Program) + grp, ctx := errgroup.WithContext(context.TODO()) + grp.SetLimit(16) + for _, rt := range routes { + if len(rt.Modules) > 0 { + route := rt + grp.Go(func() error { + mod := route.Modules[0] + var ( + err error + program *goja.Program + modPayload string = mod.Payload + ) + if mod.Type == spec.ModuleTypeTypescript { + if modPayload, err = typescript.Transpile(ctx, modPayload); err != nil { + ps.logger.Error("Error transpiling module: " + mod.Name) + return err + } } - } - if mod.Type == spec.ModuleTypeJavascript || mod.Type == spec.ModuleTypeTypescript { - if program, err = goja.Compile(mod.Name, modPayload, true); err != nil { - ps.logger.Error("Error compiling module: " + mod.Name) - return err + if mod.Type == spec.ModuleTypeJavascript || mod.Type == spec.ModuleTypeTypescript { + if program, err = goja.Compile(mod.Name, modPayload, true); err != nil { + ps.logger.Error("Error compiling module: " + mod.Name) + return err + } + } else { + return errors.New("invalid module type: " + mod.Type.String()) } - } else { - return errors.New("invalid module type: " + mod.Type.String()) - } - testRtCtx := NewRuntimeContext(ps, route, mod) - defer testRtCtx.Clean() - err = extractors.SetupModuleEventLoop(ps.printer, testRtCtx) - if err != nil { - ps.logger.Error("Error applying module changes", - zap.Error(err), zap.String("module", mod.Name), - ) - return err - } - ps.modPrograms.Insert(mod.Name+"/"+mod.Namespace.Name, program) - elapsed := time.Since(start) - ps.logger.Debug("Module changed applied", - zap.Duration("elapsed", elapsed), - zap.String("name", mod.Name), - zap.String("namespace", mod.Namespace.Name), - ) + tmpCtx := NewRuntimeContext(ps, route, mod) + defer tmpCtx.Clean() + if err = extractors.SetupModuleEventLoop(ps.printer, tmpCtx); err != nil { + ps.logger.Error("Error applying module changes", + zap.Error(err), zap.String("module", mod.Name), + ) + return err + } + programMap[mod.Name+"/"+route.Namespace.Name] = program + return nil + }) } } + + if err := grp.Wait(); err != nil { + return err + } + + for k, v := range programMap { + ps.modPrograms.Insert(k, v) + } + return nil } -func (ps *ProxyState) setupRoutes() (err error) { - ps.logger.Debug("Setting up routes") - // reqCtxProviders := avl.NewTree[string, *RequestContextProvider]() - for namespaceName, routes := range ps.rm.GetRouteNamespaceMap() { +func (ps *ProxyState) setupRoutes(log *spec.ChangeLog) (err error) { + var rtMap map[string][]*spec.DGateRoute + if log.Namespace == "" || ps.pendingChanges { + rtMap = ps.rm.GetRouteNamespaceMap() + } else { + rtMap = make(map[string][]*spec.DGateRoute) + rtMap[log.Namespace] = ps.rm.GetRoutesByNamespace(log.Namespace) + } + for namespaceName, routes := range rtMap { mux := router.NewMux() for _, rt := range routes { reqCtxProvider := NewRequestContextProvider(rt, ps) @@ -145,7 +158,6 @@ func (ps *ProxyState) setupRoutes() (err error) { }(rt) } - ps.logger.Debug("Routes have changed, reloading") if dr, ok := ps.routers.Find(namespaceName); ok { dr.ReplaceMux(mux) } else { @@ -244,11 +256,17 @@ func (ps *ProxyState) startProxyServerTLS() { } hostPort := fmt.Sprintf("%s:%d", cfg.Host, cfg.TLS.Port) ps.logger.Info("Starting secure proxy server on " + hostPort) - proxyHttpsLogger := ps.logger.Named("https") + goLogger, err := zap.NewStdLogAt( + ps.logger.Named("https"), + zap.DebugLevel, + ) + if err != nil { + panic(err) + } secureServer := &http.Server{ Addr: hostPort, Handler: ps, - ErrorLog: zap.NewStdLog(proxyHttpsLogger), + ErrorLog: goLogger, TLSConfig: ps.DynamicTLSConfig( cfg.TLS.CertFile, cfg.TLS.KeyFile, @@ -273,6 +291,7 @@ func (ps *ProxyState) startProxyServerTLS() { func (ps *ProxyState) Start() (err error) { defer func() { if err != nil { + ps.logger.Error("Error starting proxy server", zap.Error(err)) ps.Stop() } }() @@ -308,10 +327,9 @@ func (ps *ProxyState) Stop() { defer ps.Logger().Sync() ps.proxyLock.Lock() - raftNode := ps.Raft() - ps.proxyLock.Unlock() + defer ps.proxyLock.Unlock() - if raftNode != nil { + if raftNode := ps.Raft(); raftNode != nil { ps.logger.Info("Stopping Raft node") if err := raftNode.Shutdown().Error(); err != nil { ps.logger.Error("Error stopping Raft node", zap.Error(err)) diff --git a/internal/proxy/proxy_handler.go b/internal/proxy/proxy_handler.go index 8c97626..547fe2e 100644 --- a/internal/proxy/proxy_handler.go +++ b/internal/proxy/proxy_handler.go @@ -38,7 +38,7 @@ func proxyHandler(ps *ProxyState, reqCtx *RequestContext) { if reqCtx.route.Service != nil { event = event.With(zap.String("service", reqCtx.route.Service.Name)) } - event.Info("Request log") + event.Debug("Request log") }() defer ps.metrics.MeasureProxyRequest(reqCtx, time.Now()) @@ -164,7 +164,7 @@ func handleServiceProxy(ps *ProxyState, reqCtx *RequestContext, modExt ModuleExt }). ErrorHandler(func(w http.ResponseWriter, r *http.Request, reqErr error) { upstreamErr = reqErr - ps.logger.Error("Error proxying request", + ps.logger.Debug("Error proxying request", zap.String("error", reqErr.Error()), zap.String("route", reqCtx.route.Name), zap.String("service", reqCtx.route.Service.Name), diff --git a/internal/proxy/proxy_printer.go b/internal/proxy/proxy_printer.go index e37278d..ccb0878 100644 --- a/internal/proxy/proxy_printer.go +++ b/internal/proxy/proxy_printer.go @@ -1,40 +1,49 @@ package proxy -import "go.uber.org/zap" +import ( + "go.uber.org/zap" +) type ( ProxyPrinter struct { logger *zap.Logger - // logs []*printerLog } - // printerLog struct { - // time time.Time - // level string - // msg string - // } ) -func NewProxyPrinter(logger *zap.Logger) *ProxyPrinter { - return &ProxyPrinter{ - logger: logger, - // logs: make([]*printerLog, 0), +// NewProxyPrinter creates a new ProxyPrinter. +func NewProxyPrinter(logger *zap.Logger, lvl zap.AtomicLevel) *ProxyPrinter { + newLogger := logger.WithOptions(zap.IncreaseLevel(lvl)) + if !logger.Core().Enabled(lvl.Level()) { + logger.Warn("the desired log level is lower than the global log level") } + return &ProxyPrinter{newLogger} } +// Error logs a message at error level. func (pp *ProxyPrinter) Error(s string) { - // pp.logs = append(pp.logs, &printerLog{ - // time.Now(), "error", s}) pp.logger.Error(s) } +// Warn logs a message at warn level. func (pp *ProxyPrinter) Warn(s string) { - // pp.logs = append(pp.logs, &printerLog{ - // time.Now(), "warn", s}) pp.logger.Warn(s) } +// Log logs a message at debug level. func (pp *ProxyPrinter) Log(s string) { - // pp.logs = append(pp.logs, &printerLog{ - // time.Now(), "info", s}) + pp.logger.Debug(s) +} + +/* + Note: The following methods are not used but are included for completeness. +*/ + +// Info logs a message at info level. +func (pp *ProxyPrinter) Info(s string) { + pp.logger.Info(s) +} + +// Debug logs a message at debug level. +func (pp *ProxyPrinter) Debug(s string) { pp.logger.Debug(s) } diff --git a/internal/proxy/proxy_state.go b/internal/proxy/proxy_state.go index 0b0104b..44ad2d7 100644 --- a/internal/proxy/proxy_state.go +++ b/internal/proxy/proxy_state.go @@ -34,18 +34,18 @@ import ( ) type ProxyState struct { - version string - debugMode bool - changeHash uint32 - startTime time.Time - logger *zap.Logger - printer console.Printer - config *config.DGateConfig - store *proxystore.ProxyStore - changeLogs []*spec.ChangeLog - metrics *ProxyMetrics - sharedCache cache.TCache - proxyLock *sync.RWMutex + debugMode bool + changeHash *atomic.Uint64 + startTime time.Time + logger *zap.Logger + printer console.Printer + config *config.DGateConfig + store *proxystore.ProxyStore + changeLogs []*spec.ChangeLog + metrics *ProxyMetrics + sharedCache cache.TCache + proxyLock *sync.RWMutex + pendingChanges bool rm *resources.ResourceManager skdr scheduler.Scheduler @@ -53,7 +53,7 @@ type ProxyState struct { providers avl.Tree[string, *RequestContextProvider] modPrograms avl.Tree[string, *goja.Program] - ready atomic.Bool + ready *atomic.Bool replicationSettings *ProxyReplication replicationEnabled bool routers avl.Tree[string, *router.DynamicRouter] @@ -66,20 +66,21 @@ type ProxyState struct { func NewProxyState(logger *zap.Logger, conf *config.DGateConfig) *ProxyState { var dataStore storage.Storage switch conf.Storage.StorageType { - case config.StorageTypeDebug: - dataStore = storage.NewDebugStore(&storage.DebugStoreConfig{ - Logger: logger, - }) case config.StorageTypeMemory: - dataStore = storage.NewMemoryStore(&storage.MemoryStoreConfig{ - Logger: logger, - }) + memConfig, err := config.StoreConfig[storage.MemStoreConfig](conf.Storage.Config) + if err != nil { + panic(fmt.Errorf("invalid config: %s", err)) + } else { + memConfig.Logger = logger + } + dataStore = storage.NewMemStore(&memConfig) case config.StorageTypeFile: fileConfig, err := config.StoreConfig[storage.FileStoreConfig](conf.Storage.Config) if err != nil { panic(fmt.Errorf("invalid config: %s", err)) + } else { + fileConfig.Logger = logger } - fileConfig.Logger = logger dataStore = storage.NewFileStore(&fileConfig) default: panic(fmt.Errorf("invalid storage type: %s", conf.Storage.StorageType)) @@ -91,9 +92,11 @@ func NewProxyState(logger *zap.Logger, conf *config.DGateConfig) *ProxyState { opt = resources.WithDefaultNamespace(spec.DefaultNamespace) } var printer console.Printer = &extractors.NoopPrinter{} - if conf.ProxyConfig.EnableConsoleLogger { - printer = NewProxyPrinter(logger) + consoleLevel, err := zap.ParseAtomicLevel(conf.ProxyConfig.ConsoleLogLevel) + if err != nil { + panic(fmt.Errorf("invalid console log level: %s", err)) } + printer = NewProxyPrinter(logger, consoleLevel) rpLogger := logger.Named("reverse-proxy") storeLogger := logger.Named("store") schedulerLogger := logger.Named("scheduler") @@ -103,15 +106,16 @@ func NewProxyState(logger *zap.Logger, conf *config.DGateConfig) *ProxyState { replicationEnabled = true } state := &ProxyState{ - startTime: time.Now(), - ready: atomic.Bool{}, - logger: logger, - debugMode: conf.Debug, - config: conf, - metrics: NewProxyMetrics(), - printer: printer, - routers: avl.NewTree[string, *router.DynamicRouter](), - rm: resources.NewManager(opt), + startTime: time.Now(), + ready: new(atomic.Bool), + changeHash: new(atomic.Uint64), + logger: logger, + debugMode: conf.Debug, + config: conf, + metrics: NewProxyMetrics(), + printer: printer, + routers: avl.NewTree[string, *router.DynamicRouter](), + rm: resources.NewManager(opt), skdr: scheduler.New(scheduler.Options{ Logger: schedulerLogger, }), @@ -150,10 +154,6 @@ func NewProxyState(logger *zap.Logger, conf *config.DGateConfig) *ProxyState { return state } -func (ps *ProxyState) Version() string { - return ps.version -} - func (ps *ProxyState) Store() *proxystore.ProxyStore { return ps.store } @@ -162,19 +162,19 @@ func (ps *ProxyState) Logger() *zap.Logger { return ps.logger } -func (ps *ProxyState) ChangeHash() uint32 { - return ps.changeHash +func (ps *ProxyState) ChangeHash() uint64 { + return ps.changeHash.Load() } func (ps *ProxyState) ChangeLogs() []*spec.ChangeLog { - return ps.changeLogs + // return a copy of the change logs + ps.proxyLock.RLock() + defer ps.proxyLock.RUnlock() + return append([]*spec.ChangeLog{}, ps.changeLogs...) } func (ps *ProxyState) Ready() bool { - if ps.replicationEnabled { - return ps.ready.Load() - } - return true + return ps.ready.Load() } func (ps *ProxyState) Raft() *raft.Raft { @@ -184,76 +184,77 @@ func (ps *ProxyState) Raft() *raft.Raft { return nil } -func (ps *ProxyState) SetupRaft(r *raft.Raft, oc chan raft.Observation) { +func (ps *ProxyState) SetupRaft(r *raft.Raft) { ps.proxyLock.Lock() defer ps.proxyLock.Unlock() + + oc := make(chan raft.Observation, 32) + r.RegisterObserver(raft.NewObserver(oc, true, nil)) go func() { + logger := ps.logger.Named("raft-spy") for obs := range oc { - switch raftObs := obs.Data.(type) { + switch ro := obs.Data.(type) { case raft.PeerObservation: - ps.logger.Info("peer observation", - zap.Stringer("suffrage", raftObs.Peer.Suffrage), - zap.String("address", string(raftObs.Peer.Address)), - zap.String("id", string(raftObs.Peer.ID)), + logger.Info("peer observation", + zap.Stringer("suffrage", ro.Peer.Suffrage), + zap.String("address", string(ro.Peer.Address)), + zap.String("id", string(ro.Peer.ID)), ) case raft.LeaderObservation: - ps.logger.Info("leader observation", - zap.String("leader_addr", string(raftObs.LeaderAddr)), - zap.String("leader_id", string(raftObs.LeaderID)), + logger.Info("leader observation", + zap.String("leader_addr", string(ro.LeaderAddr)), + zap.String("leader_id", string(ro.LeaderID)), ) case raft.RequestVoteRequest: - ps.logger.Info("request vote request", - zap.String("candidate_id", string(raftObs.GetRPCHeader().ID)), - zap.String("candidate_addr", string(raftObs.GetRPCHeader().Addr)), - zap.Uint64("term", raftObs.Term), - zap.Uint64("last-log-index", raftObs.LastLogIndex), - zap.Uint64("last-log-term", raftObs.LastLogTerm), + logger.Info("request vote request", + zap.String("candidate_id", string(ro.GetRPCHeader().ID)), + zap.String("candidate_addr", string(ro.GetRPCHeader().Addr)), + zap.Uint64("term", ro.Term), + zap.Uint64("last-log-index", ro.LastLogIndex), + zap.Uint64("last-log-term", ro.LastLogTerm), ) } } - + panic("raft observer channel closed") }() - ps.replicationSettings = NewProxyReplication(r) } func (ps *ProxyState) WaitForChanges() error { - if rft := ps.Raft(); rft != nil { - return rft.Barrier(time.Second * 5).Error() - } else { - ps.proxyLock.RLock() - defer ps.proxyLock.RUnlock() + ps.proxyLock.RLock() + defer ps.proxyLock.RUnlock() + if r := ps.Raft(); r != nil && r.State() == raft.Leader { + return r.Barrier(time.Second * 5).Error() } return nil } func (ps *ProxyState) ApplyChangeLog(log *spec.ChangeLog) error { - if ps.replicationEnabled { - if log.Cmd.IsNoop() { - return ps.processChangeLog(log, true, false) - } - r := ps.replicationSettings.raft + if r := ps.Raft(); r != nil { if r.State() != raft.Leader { return raft.ErrNotLeader } - encodedCL, err := json.Marshal(log) - if err != nil { + if err := ps.processChangeLog(log, true, false); err != nil { return err } - raftLog := raft.Log{ - Data: encodedCL, - } - err = ps.ProcessChangeLog(log, true) + encodedCL, err := json.Marshal(log) if err != nil { return err } + raftLog := raft.Log{Data: encodedCL} + now := time.Now() future := r.ApplyLog(raftLog, time.Second*15) + err = future.Error() ps.logger.With(). Debug("waiting for reply from raft", zap.String("id", log.ID), zap.Stringer("command", log.Cmd), + zap.Stringer("command", time.Since(now)), + zap.Uint64("index", future.Index()), + zap.Any("response", future.Response()), + zap.Error(err), ) - return future.Error() + return err } else { return ps.processChangeLog(log, true, true) } @@ -296,10 +297,11 @@ func (ps *ProxyState) restartState(fn func(error)) { fn(err) return } - } - if err := ps.restoreFromChangeLogs(true); err != nil { - fn(err) - return + } else { + if err := ps.restoreFromChangeLogs(true); err != nil { + fn(err) + return + } } ps.logger.Info("State successfully restarted") fn(nil) @@ -324,8 +326,7 @@ func (ps *ProxyState) ReloadState(check bool, logs ...*spec.ChangeLog) error { } func (ps *ProxyState) ProcessChangeLog(log *spec.ChangeLog, reload bool) error { - err := ps.processChangeLog(log, reload, !ps.replicationEnabled) - if err != nil { + if err := ps.processChangeLog(log, reload, true); err != nil { ps.logger.Error("processing error", zap.Error(err)) return err } @@ -578,7 +579,7 @@ func (ps *ProxyState) ServeHTTP(w http.ResponseWriter, r *http.Request) { // if debug mode is enabled, return a 403 util.WriteStatusCodeError(w, http.StatusForbidden) if ps.debugMode { - w.Write([]byte(" - Domain not allowed")) + w.Write([]byte("domain not allowed")) } return } diff --git a/internal/proxy/proxystore/proxy_store.go b/internal/proxy/proxystore/proxy_store.go index d55eb9a..6439e32 100644 --- a/internal/proxy/proxystore/proxy_store.go +++ b/internal/proxy/proxystore/proxy_store.go @@ -32,6 +32,14 @@ func (store *ProxyStore) InitStore() error { return nil } +func (store *ProxyStore) CloseStore() error { + err := store.storage.Close() + if err != nil { + return err + } + return nil +} + func (store *ProxyStore) FetchChangeLogs() ([]*spec.ChangeLog, error) { clBytes, err := store.storage.GetPrefix("changelog/", 0, -1) if err != nil { @@ -144,7 +152,6 @@ func (store *ProxyStore) StoreDocument(doc *spec.Document) error { if err != nil { return err } - store.logger.Debug("storing document") err = store.storage.Set(createDocumentKey(doc.ID, doc.CollectionName, doc.NamespaceName), docBytes) if err != nil { return err @@ -164,7 +171,11 @@ func (store *ProxyStore) DeleteDocument(id, colName, nsName string) error { } func (store *ProxyStore) DeleteDocuments(doc *spec.Document) error { - err := store.storage.IterateTxnPrefix(createDocumentKey("", doc.CollectionName, doc.NamespaceName), + docKey := createDocumentKey( + "", doc.CollectionName, + doc.NamespaceName, + ) + err := store.storage.IterateTxnPrefix(docKey, func(txn storage.StorageTxn, key string) error { return txn.Delete(key) }) diff --git a/internal/proxy/runtime_context.go b/internal/proxy/runtime_context.go index 0c901b8..94c52db 100644 --- a/internal/proxy/runtime_context.go +++ b/internal/proxy/runtime_context.go @@ -25,6 +25,8 @@ type runtimeContext struct { modules []*spec.Module } +var _ modules.RuntimeContext = &runtimeContext{} + func NewRuntimeContext( proxyState *ProxyState, route *spec.DGateRoute, @@ -39,6 +41,7 @@ func NewRuntimeContext( reg := require.NewRegistryWithLoader(func(path string) ([]byte, error) { requireMod := strings.Replace(path, "node_modules/", "", 1) + // TODO: add support for other module types w/ permissions // 'https://' - requires network permissions and must be enabled in the config // 'file://' - requires file system permissions and must be enabled in the config // 'module://' - requires a module lookup and module permissions @@ -57,7 +60,8 @@ func NewRuntimeContext( return code.([]byte), nil } } - payload, err := typescript.Transpile(mod.Payload) + payload, err := typescript.Transpile( + context.TODO(), mod.Payload) if err != nil { return nil, err } @@ -71,8 +75,6 @@ func NewRuntimeContext( return rtCtx } -var _ modules.RuntimeContext = &runtimeContext{} - // UseRequestContext sets the request context func (rtCtx *runtimeContext) Use(reqCtx *RequestContext) (*runtimeContext, error) { if reqCtx != nil { diff --git a/internal/proxy/util.go b/internal/proxy/util.go index af8856e..2ffde8d 100644 --- a/internal/proxy/util.go +++ b/internal/proxy/util.go @@ -7,14 +7,14 @@ import ( "encoding/json" "errors" "hash" - "hash/crc32" + "hash/crc64" "net/http" "slices" "sort" ) -func saltHash[T any](salt uint32, objs ...T) (hash.Hash32, error) { - hash := crc32.NewIEEE() +func saltHash[T any](salt uint64, objs ...T) (hash.Hash64, error) { + hash := crc64.New(crc64.MakeTable(crc64.ECMA)) if salt != 0 { // uint32 to byte array b := make([]byte, 4) @@ -39,15 +39,15 @@ func saltHash[T any](salt uint32, objs ...T) (hash.Hash32, error) { return hash, nil } -func HashAny[T any](salt uint32, objs ...T) (uint32, error) { +func HashAny[T any](salt uint64, objs ...T) (uint64, error) { h, err := saltHash(salt, objs...) if err != nil { return 0, err } - return h.Sum32(), nil + return h.Sum64(), nil } -func HashString[T any](salt uint32, objs ...T) (string, error) { +func HashString[T any](salt uint64, objs ...T) (string, error) { h, err := saltHash(salt, objs...) if err != nil { return "", err diff --git a/performance-tests/long-perf-test.js b/performance-tests/long-perf-test.js index 78cf7ab..c835d21 100644 --- a/performance-tests/long-perf-test.js +++ b/performance-tests/long-perf-test.js @@ -91,7 +91,7 @@ export function dgatePath() { const dgatePath = __ENV._PROXY_URL || 'http://localhost' const path = __ENV.DGATE_PATH; let res = http.get(dgatePath + path, { - headers: { Host: 'dgate.dev' }, + headers: { Host: 'performance.example.com' }, }); let results = {}; results[path + ': status is ' + res.status] = diff --git a/performance-tests/perf-test.js b/performance-tests/perf-test.js index a906bc5..69b5c46 100644 --- a/performance-tests/perf-test.js +++ b/performance-tests/perf-test.js @@ -40,7 +40,7 @@ export function dgatePath() { const dgatePath = __ENV.PROXY_URL || 'http://localhost'; const path = __ENV.DGATE_PATH; let res = http.get(dgatePath + path, { - headers: { Host: 'dgate.dev' }, + headers: { Host: 'performance.example.com' }, }); let results = {}; results[path + ': status is ' + res.status] = (r) => r.status < 400; diff --git a/pkg/modules/extractors/async_tracker.go b/pkg/modules/extractors/async_tracker.go deleted file mode 100644 index c158f43..0000000 --- a/pkg/modules/extractors/async_tracker.go +++ /dev/null @@ -1,66 +0,0 @@ -package extractors - -import ( - "context" - "fmt" - "sync/atomic" - - "github.com/dop251/goja" -) - -var _ goja.AsyncContextTracker = &asyncTracker{} - -type asyncTracker struct { - count atomic.Int32 - exitChan chan int32 -} - -type TrackerEvent int - -const ( - Exited TrackerEvent = iota - Resumed -) - -func newAsyncTracker() *asyncTracker { - return &asyncTracker{ - count: atomic.Int32{}, - exitChan: make(chan int32, 128), - } -} - -// Exited is called when an async function is done -func (t *asyncTracker) Exited() { - t.exitChan <- t.count.Add(-1) -} - -// Grab is called when an async function is scheduled -func (t *asyncTracker) Grab() any { - t.exitChan <- t.count.Add(1) - return nil -} - -// Resumed is called when an async function is executed (ignore) -func (t *asyncTracker) Resumed(any) { - t.exitChan <- t.count.Load() -} - -func (t *asyncTracker) waitTimeout( - ctx context.Context, doneFn func() bool, -) error { - if doneFn() { - return nil - } else if t.count.Load() == 0 { - return nil - } - for { - select { - case <-ctx.Done(): - return fmt.Errorf("async tracker: %s", ctx.Err()) - case numLeft := <-t.exitChan: - if numLeft == 0 || doneFn() { - return nil - } - } - } -} diff --git a/pkg/modules/extractors/extractors.go b/pkg/modules/extractors/extractors.go index a52152d..cdfa54e 100644 --- a/pkg/modules/extractors/extractors.go +++ b/pkg/modules/extractors/extractors.go @@ -45,28 +45,20 @@ func RunAndWaitForResult( fn goja.Callable, args ...goja.Value, ) (res goja.Value, err error) { - tracker := newAsyncTracker() - rt.SetAsyncContextTracker(tracker) - defer func() { - rt.SetAsyncContextTracker(nil) - if err != nil { - rt.Interrupt(err.Error()) - } - }() - if res, err = fn(nil, args...); err != nil { return nil, err } else if prom, ok := res.Export().(*goja.Promise); ok { ctx, cancel := context.WithTimeout( - context.TODO(), 30*time.Second, - ) + context.TODO(), 30*time.Second) defer cancel() - if err := tracker.waitTimeout(ctx, func() bool { + if err = waitTimeout(ctx, func() bool { return prom.State() != goja.PromiseStatePending }); err != nil { + rt.Interrupt(err.Error()) return nil, errors.New("promise timed out: " + err.Error()) } if prom.State() == goja.PromiseStateRejected { + // no need to interrupt the runtime here return nil, errors.New(prom.Result().String()) } results := prom.Result() @@ -85,6 +77,29 @@ func nully(val goja.Value) bool { return val == nil || goja.IsUndefined(val) || goja.IsNull(val) } +func waitTimeout(ctx context.Context, doneFn func() bool) error { + if doneFn() { + return nil + } + maxTimeout := 100 * time.Millisecond + multiplier := 1.75 + backoffTimeout := 2 * time.Millisecond + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoffTimeout): + if !doneFn() { + backoffTimeout = min(time.Duration( + float64(backoffTimeout)*multiplier, + ), maxTimeout) + continue + } + return nil + } + } +} + func DefaultFetchUpstreamFunction() FetchUpstreamUrlFunc { roundRobinIndex := 0 return func(ctx *types.ModuleContext) (*url.URL, error) { diff --git a/pkg/modules/extractors/extractors_test.go b/pkg/modules/extractors/extractors_test.go index 71001ba..a58dde2 100644 --- a/pkg/modules/extractors/extractors_test.go +++ b/pkg/modules/extractors/extractors_test.go @@ -1,6 +1,7 @@ package extractors_test import ( + "context" "strconv" "testing" @@ -39,7 +40,7 @@ async function print() {console.log("log")} ` func Test_runAndWaitForResult(t *testing.T) { - src, err := typescript.Transpile(TS_PAYLOAD) + src, err := typescript.Transpile(context.Background(), TS_PAYLOAD) if err != nil { t.Fatal(err) } @@ -96,7 +97,7 @@ export async function named_func_async() { ` func TestExportedInformation(t *testing.T) { - src, err := typescript.Transpile(TS_PAYLOAD_EXPORTED) + src, err := typescript.Transpile(context.Background(), TS_PAYLOAD_EXPORTED) if err != nil { t.Fatal(err) } @@ -172,7 +173,7 @@ export async function test2() { ` func TestExportedPromiseErrors(t *testing.T) { - src, err := typescript.Transpile(TS_PAYLOAD_PROMISE) + src, err := typescript.Transpile(context.Background(), TS_PAYLOAD_PROMISE) if err != nil { t.Fatal(err) } diff --git a/pkg/modules/extractors/runtime_test.go b/pkg/modules/extractors/runtime_test.go index 068d812..324f537 100644 --- a/pkg/modules/extractors/runtime_test.go +++ b/pkg/modules/extractors/runtime_test.go @@ -1,6 +1,7 @@ package extractors_test import ( + "context" "testing" "github.com/dgate-io/dgate/internal/config/configtest" @@ -147,7 +148,7 @@ func BenchmarkNewModuleRuntime(b *testing.B) { b.Run("Transpile-TS", func(b *testing.B) { for i := 0; i < b.N; i++ { b.StartTimer() - _, err := typescript.Transpile(TS_PAYLOAD_CUSTOMFUNC) + _, err := typescript.Transpile(context.Background(), TS_PAYLOAD_CUSTOMFUNC) if err != nil { b.Fatal(err) } diff --git a/pkg/modules/testutil/testutil.go b/pkg/modules/testutil/testutil.go index 97a97cf..70ad006 100644 --- a/pkg/modules/testutil/testutil.go +++ b/pkg/modules/testutil/testutil.go @@ -127,7 +127,7 @@ type Crashable interface { } func CreateTSProgram(c Crashable, payload string) *goja.Program { - src, err := typescript.Transpile(payload) + src, err := typescript.Transpile(context.Background(), payload) if err != nil { c.Fatal(err) } diff --git a/pkg/rafthttp/rafthttp.go b/pkg/rafthttp/rafthttp.go index d01ea35..af28bdc 100644 --- a/pkg/rafthttp/rafthttp.go +++ b/pkg/rafthttp/rafthttp.go @@ -36,6 +36,7 @@ type HTTPTransport struct { } var _ raft.Transport = (*HTTPTransport)(nil) +var _ raft.WithPreVote = (*HTTPTransport)(nil) func NewHTTPTransport(addr raft.ServerAddress, client Doer, logger *zap.Logger, urlFmt string) *HTTPTransport { if client == nil { @@ -132,6 +133,11 @@ func (t *HTTPTransport) RequestVote(_ raft.ServerID, target raft.ServerAddress, return t.send(t.generateUrl(target, "RequestVote"), args, resp) } +// RequestPreVote implements the raft.Transport interface. +func (t *HTTPTransport) RequestPreVote(_ raft.ServerID, target raft.ServerAddress, args *raft.RequestPreVoteRequest, resp *raft.RequestPreVoteResponse) error { + return t.send(t.generateUrl(target, "RequestPreVote"), args, resp) +} + // InstallSnapshot implements the raft.Transport interface. func (t *HTTPTransport) InstallSnapshot(_ raft.ServerID, target raft.ServerAddress, args *raft.InstallSnapshotRequest, resp *raft.InstallSnapshotResponse, data io.Reader) error { // Send a dummy request to see if the remote host supports @@ -280,6 +286,8 @@ func (t *HTTPTransport) ServeHTTP(res http.ResponseWriter, req *http.Request) { return case "RequestVote": rpc.Command = &raft.RequestVoteRequest{} + case "RequestPreVote": + rpc.Command = &raft.RequestPreVoteRequest{} case "AppendEntries": rpc.Command = &raft.AppendEntriesRequest{} case "TimeoutNow": diff --git a/pkg/storage/debug_storage.go b/pkg/storage/debug_storage.go deleted file mode 100644 index e74eb0b..0000000 --- a/pkg/storage/debug_storage.go +++ /dev/null @@ -1,93 +0,0 @@ -package storage - -import ( - "errors" - "strings" - - "github.com/dgate-io/dgate/pkg/util/tree/avl" - "go.uber.org/zap" -) - -type DebugStoreConfig struct { - Logger *zap.Logger -} - -type DebugStore struct { - tree avl.Tree[string, []byte] -} - -var _ Storage = &DebugStore{} - -func NewDebugStore(cfg *DebugStoreConfig) *DebugStore { - return &DebugStore{ - tree: avl.NewTree[string, []byte](), - } -} - -func (m *DebugStore) Connect() error { - return nil -} - -func (m *DebugStore) Get(key string) ([]byte, error) { - if b, ok := m.tree.Find(key); ok { - return b, nil - } - return nil, errors.New("key not found") -} - -func (m *DebugStore) Set(key string, value []byte) error { - m.tree.Insert(key, value) - return nil -} - -func (m *DebugStore) IterateValuesPrefix(prefix string, fn func(string, []byte) error) error { - check := true - m.tree.Each(func(k string, v []byte) bool { - if strings.HasPrefix(k, prefix) { - check = true - if err := fn(k, v); err != nil { - return false - } - return true - } - return check - }) - return nil -} - -func (m *DebugStore) IterateTxnPrefix(prefix string, fn func(StorageTxn, string) error) error { - panic("implement me") -} - -func (m *DebugStore) GetPrefix(prefix string, offset, limit int) ([]*KeyValue, error) { - if limit <= 0 { - limit = 0 - } - kvs := make([]*KeyValue, 0, limit) - m.IterateValuesPrefix(prefix, func(key string, value []byte) error { - if offset <= 0 { - if len(kvs) >= limit { - return errors.New("limit reached") - } - kvs = append(kvs, &KeyValue{ - Key: key, - Value: value, - }) - } else { - offset-- - } - return nil - }) - return kvs, nil -} - -func (m *DebugStore) Delete(key string) error { - if ok := m.tree.Delete(key); !ok { - return errors.New("key not found") - } - return nil -} - -func (m *DebugStore) Close() error { - return nil -} diff --git a/pkg/storage/file_storage.go b/pkg/storage/file_storage.go index 3fd417b..ccf9227 100644 --- a/pkg/storage/file_storage.go +++ b/pkg/storage/file_storage.go @@ -1,13 +1,13 @@ package storage import ( + "bytes" "errors" "fmt" - "os" + "path" "strings" - "github.com/dgraph-io/badger/v4" - "github.com/dgraph-io/badger/v4/options" + bolt "go.etcd.io/bbolt" "go.uber.org/zap" ) @@ -17,19 +17,20 @@ type FileStoreConfig struct { } type FileStore struct { - directory string - logger badger.Logger - inMemory bool - db *badger.DB + logger *zap.Logger + bucketName []byte + directory string + db *bolt.DB } type FileStoreTxn struct { - txn *badger.Txn - ro bool + txn *bolt.Tx + ro bool + bucket *bolt.Bucket } -var _ Storage = &FileStore{} -var _ StorageTxn = &FileStoreTxn{} +var _ Storage = (*FileStore)(nil) +var _ StorageTxn = (*FileStoreTxn)(nil) var ( // ErrStoreLocked is returned when the storage is locked. @@ -51,115 +52,88 @@ func NewFileStore(fsConfig *FileStoreConfig) *FileStore { fsConfig.Directory = strings.TrimSuffix(fsConfig.Directory, "/") } - return &FileStore{ - directory: fsConfig.Directory, - logger: newBadgerLoggerAdapter( - "filestore::badger", fsConfig.Logger, - ), - inMemory: false, + if fsConfig.Logger == nil { + fsConfig.Logger = zap.NewNop() } -} -func newFileStoreTxn(txn *badger.Txn) *FileStoreTxn { - return &FileStoreTxn{ - txn: txn, + return &FileStore{ + directory: fsConfig.Directory, + logger: fsConfig.Logger.Named("boltstore::bolt"), + bucketName: []byte("dgate"), } } -func (s *FileStore) Connect() error { - var opts badger.Options - var err error - if s.inMemory { - opts = badger.DefaultOptions(""). - WithCompression(options.Snappy). - WithInMemory(true). - WithLogger(s.logger) +func (s *FileStore) Connect() (err error) { + filePath := path.Join(s.directory, "dgate.db") + if s.db, err = bolt.Open(filePath, + 0755, bolt.DefaultOptions, + ); err != nil { + return err + } + if tx, err := s.db.Begin(true); err != nil { + return err } else { - // Create the directory if it does not exist. - if _, err := os.Stat(s.directory); os.IsNotExist(err) { - err := os.MkdirAll(s.directory, 0755) - if err != nil { - return errors.New("failed to create directory - " + s.directory + ": " + err.Error()) - } + _, err = tx.CreateBucketIfNotExists(s.bucketName) + if err != nil { + return err } - - opts = badger.DefaultOptions(s.directory). - WithReadOnly(false). - WithInMemory(s.inMemory). - WithCompression(options.Snappy). - WithLogger(s.logger) + return tx.Commit() } - s.db, err = badger.Open(opts) - if err != nil { - return err +} + +func (s *FileStore) newTxn(txn *bolt.Tx) *FileStoreTxn { + if bucket := txn.Bucket(s.bucketName); bucket != nil { + return &FileStoreTxn{ + txn: txn, + bucket: bucket, + } } - return nil + panic("bucket not found") } func (s *FileStore) Get(key string) ([]byte, error) { var value []byte - err := s.db.View(func(txn *badger.Txn) error { - val, err := newFileStoreTxn(txn).Get(key) - if err != nil { - if err == badger.ErrKeyNotFound { - return ErrKeyNotFound - } - return err - } - value = val - return nil + return value, s.db.View(func(txn *bolt.Tx) (err error) { + value, err = s.newTxn(txn).Get(key) + return err }) - return value, err } func (s *FileStore) Set(key string, value []byte) error { - return s.db.Update(func(txn *badger.Txn) error { - return newFileStoreTxn(txn).Set(key, value) + return s.db.Update(func(txn *bolt.Tx) error { + return s.newTxn(txn).Set(key, value) + }) +} + +func (s *FileStore) Delete(key string) error { + return s.db.Update(func(txn *bolt.Tx) error { + return s.newTxn(txn).Delete(key) }) } func (s *FileStore) IterateValuesPrefix(prefix string, fn func(string, []byte) error) error { - return s.db.View(func(txn *badger.Txn) error { - return newFileStoreTxn(txn).IterateValuesPrefix(prefix, fn) + return s.db.View(func(txn *bolt.Tx) error { + return s.newTxn(txn).IterateValuesPrefix(prefix, fn) }) } func (s *FileStore) IterateTxnPrefix(prefix string, fn func(StorageTxn, string) error) error { - return s.db.View(func(txn *badger.Txn) error { - return newFileStoreTxn(txn).IterateTxnPrefix(prefix, fn) + return s.db.Update(func(txn *bolt.Tx) error { + return s.newTxn(txn).IterateTxnPrefix(prefix, fn) }) } func (s *FileStore) GetPrefix(prefix string, offset, limit int) ([]*KeyValue, error) { - var return_list []*KeyValue - err := s.db.View(func(txn *badger.Txn) error { - val, err := newFileStoreTxn(txn).GetPrefix(prefix, offset, limit) + var list []*KeyValue + err := s.db.View(func(txn *bolt.Tx) error { + val, err := s.newTxn(txn).GetPrefix(prefix, offset, limit) if err != nil { return fmt.Errorf("failed to get prefix: %w", err) } - return_list = val + list = val return nil }) - return return_list, err -} - -func (s *FileStore) Delete(key string) error { - return s.db.Update(func(txn *badger.Txn) error { - return txn.Delete([]byte(key)) - }) -} - -func (s *FileStore) Txn(readOnly bool, fn func(StorageTxn) error) error { - txFunc := s.db.View - if !readOnly { - txFunc = s.db.Update - } - return txFunc(func(txn *badger.Txn) error { - return fn(&FileStoreTxn{ - txn: txn, - ro: readOnly, - }) - }) + return list, err } func (s *FileStore) Close() error { @@ -167,43 +141,28 @@ func (s *FileStore) Close() error { } func (tx *FileStoreTxn) Get(key string) ([]byte, error) { - item, err := tx.txn.Get([]byte(key)) - if err != nil { - return nil, err - } - val, err := item.ValueCopy(nil) - if err != nil { - return nil, err - } - return val, nil + return tx.bucket.Get([]byte(key)), nil } func (tx *FileStoreTxn) Set(key string, value []byte) error { if tx.ro { return ErrTxnReadOnly } - return tx.txn.Set([]byte(key), value) + return tx.bucket.Put([]byte(key), value) } func (tx *FileStoreTxn) Delete(key string) error { if tx.ro { return ErrTxnReadOnly } - return tx.txn.Delete([]byte(key)) + return tx.bucket.Delete([]byte(key)) } func (tx *FileStoreTxn) IterateValuesPrefix(prefix string, fn func(string, []byte) error) error { - iter := tx.txn.NewIterator(badger.IteratorOptions{ - Prefix: []byte(prefix), - }) - defer iter.Close() - for iter.Rewind(); iter.Valid(); iter.Next() { - item := iter.Item() - val, err := item.ValueCopy(nil) - if err != nil { - return err - } - if err := fn(string(item.Key()), val); err != nil { + c := tx.bucket.Cursor() + pre := []byte(prefix) + for k, v := c.Seek(pre); bytes.HasPrefix(k, pre); k, v = c.Next() { + if err := fn(string(k), v); err != nil { return err } } @@ -211,13 +170,10 @@ func (tx *FileStoreTxn) IterateValuesPrefix(prefix string, fn func(string, []byt } func (tx *FileStoreTxn) IterateTxnPrefix(prefix string, fn func(StorageTxn, string) error) error { - iter := tx.txn.NewIterator(badger.IteratorOptions{ - Prefix: []byte(prefix), - }) - defer iter.Close() - for iter.Rewind(); iter.Valid(); iter.Next() { - item := iter.Item() - if err := fn(tx, string(item.Key())); err != nil { + c := tx.bucket.Cursor() + pre := []byte(prefix) + for k, _ := c.Seek(pre); bytes.HasPrefix(k, pre); k, _ = c.Next() { + if err := fn(tx, string(k)); err != nil { return err } } @@ -225,29 +181,19 @@ func (tx *FileStoreTxn) IterateTxnPrefix(prefix string, fn func(StorageTxn, stri } func (s *FileStoreTxn) GetPrefix(prefix string, offset, limit int) ([]*KeyValue, error) { - return_list := make([]*KeyValue, 0) - iter := s.txn.NewIterator(badger.IteratorOptions{ - Prefix: []byte(prefix), - }) - defer iter.Close() - for iter.Rewind(); iter.Valid(); iter.Next() { + list := make([]*KeyValue, 0) + c := s.bucket.Cursor() + pre := []byte(prefix) + for k, v := c.Seek(pre); bytes.HasPrefix(k, pre); k, v = c.Next() { if offset > 0 { - offset -= 1 + offset-- continue } - item := iter.Item() - val, err := item.ValueCopy(nil) - if err != nil { - return nil, fmt.Errorf("error copying value: %v", err) - } - return_list = append(return_list, &KeyValue{ - Key: string(item.Key()), - Value: val, - }) - if limit -= 1; limit == 0 { + if limit == 0 { break } + list = append(list, &KeyValue{Key: string(k), Value: v}) + limit-- } - - return return_list, nil + return list, nil } diff --git a/pkg/storage/mem_storage.go b/pkg/storage/mem_storage.go new file mode 100644 index 0000000..32fc147 --- /dev/null +++ b/pkg/storage/mem_storage.go @@ -0,0 +1,133 @@ +package storage + +import ( + "errors" + "strings" + + "github.com/dgate-io/dgate/pkg/util/tree/avl" + "go.uber.org/zap" +) + +type MemStoreConfig struct { + Logger *zap.Logger +} + +type MemStore struct { + tree avl.Tree[string, []byte] +} + +type MemStoreTxn struct { + store *MemStore +} + +var _ Storage = &MemStore{} +var _ StorageTxn = &MemStoreTxn{} + +func NewMemStore(cfg *MemStoreConfig) *MemStore { + return &MemStore{ + tree: avl.NewTree[string, []byte](), + } +} + +func (m *MemStore) Connect() error { + return nil +} + +func (m *MemStore) Get(key string) ([]byte, error) { + if b, ok := m.tree.Find(key); ok { + return b, nil + } + return nil, errors.New("key not found") +} + +func (m *MemStore) Set(key string, value []byte) error { + m.tree.Insert(key, value) + return nil +} + +func (m *MemStore) IterateValuesPrefix(prefix string, fn func(string, []byte) error) error { + check := true + m.tree.Each(func(k string, v []byte) bool { + if strings.HasPrefix(k, prefix) { + check = true + if err := fn(k, v); err != nil { + return false + } + return true + } + return check + }) + return nil +} + +func (m *MemStore) IterateTxnPrefix(prefix string, fn func(StorageTxn, string) error) error { + m.tree.Each(func(k string, v []byte) bool { + if strings.HasPrefix(k, prefix) { + txn := &MemStoreTxn{ + store: m, + } + if err := fn(txn, k); err != nil { + return false + } + } + return true + }) + return nil +} + +func (m *MemStore) GetPrefix(prefix string, offset, limit int) ([]*KeyValue, error) { + if limit <= 0 { + limit = 0 + } + kvs := make([]*KeyValue, 0, limit) + m.IterateValuesPrefix(prefix, func(key string, value []byte) error { + if offset <= 0 { + if len(kvs) >= limit { + return errors.New("limit reached") + } + kvs = append(kvs, &KeyValue{ + Key: key, + Value: value, + }) + } else { + offset-- + } + return nil + }) + return kvs, nil +} + +func (m *MemStore) Delete(key string) error { + if ok := m.tree.Delete(key); !ok { + return errors.New("key not found") + } + return nil +} + +func (m *MemStore) Close() error { + return nil +} + +func (t *MemStoreTxn) Get(key string) ([]byte, error) { + return t.store.Get(key) +} + +func (t *MemStoreTxn) Set(key string, value []byte) error { + return t.store.Set(key, value) +} + +func (t *MemStoreTxn) Delete(key string) error { + return t.store.Delete(key) +} + +func (t *MemStoreTxn) GetPrefix(prefix string, offset int, limit int) ([]*KeyValue, error) { + return t.store.GetPrefix(prefix, offset, limit) +} + +func (t *MemStoreTxn) IterateTxnPrefix(prefix string, fn func(txn StorageTxn, key string) error) error { + return t.store.IterateTxnPrefix(prefix, fn) +} + +func (t *MemStoreTxn) IterateValuesPrefix(prefix string, fn func(key string, val []byte) error) error { + return t.store.IterateValuesPrefix(prefix, fn) +} diff --git a/pkg/storage/memory_storage.go b/pkg/storage/memory_storage.go deleted file mode 100644 index 722e774..0000000 --- a/pkg/storage/memory_storage.go +++ /dev/null @@ -1,27 +0,0 @@ -package storage - -import "go.uber.org/zap" - -type MemoryStoreConfig struct { - // Path to the directory where the files will be stored. - // If the directory does not exist, it will be created. - // If the directory exists, it will be used. - Logger *zap.Logger -} - -type MemoryStore struct { - *FileStore -} - -var _ Storage = &MemoryStore{} - -func NewMemoryStore(cfg *MemoryStoreConfig) *MemoryStore { - return &MemoryStore{ - FileStore: &FileStore{ - inMemory: true, - logger: newBadgerLoggerAdapter( - "memstore::badger", cfg.Logger, - ), - }, - } -} diff --git a/pkg/typescript/typescript.go b/pkg/typescript/typescript.go index 3b020b2..c0f9838 100644 --- a/pkg/typescript/typescript.go +++ b/pkg/typescript/typescript.go @@ -1,7 +1,9 @@ package typescript import ( + "context" _ "embed" + "strings" "github.com/clarkmcc/go-typescript" "github.com/dop251/goja" @@ -12,9 +14,11 @@ import ( //go:embed typescript.min.js var tscSource string -func Transpile(src string) (string, error) { +func Transpile(ctx context.Context, src string) (string, error) { + srcReader := strings.NewReader(src) // transpiles TS into JS with commonjs module and targets es5 - return typescript.TranspileString(src, + return typescript.TranspileCtx( + ctx, srcReader, WithCachedTypescriptSource(), typescript.WithPreventCancellation(), typescript.WithCompileOptions(map[string]any{ diff --git a/pkg/typescript/typescript_test.go b/pkg/typescript/typescript_test.go index 7769477..a262fc7 100644 --- a/pkg/typescript/typescript_test.go +++ b/pkg/typescript/typescript_test.go @@ -1,6 +1,7 @@ package typescript_test import ( + "context" "strings" "testing" @@ -37,7 +38,7 @@ func TestTranspile(t *testing.T) { for _, tsSrc := range tsSrcList { vm := goja.New() - jsSrc, err := typescript.Transpile(tsSrc) + jsSrc, err := typescript.Transpile(context.Background(), tsSrc) if err != nil { t.Fatal(err) return diff --git a/pkg/util/http_test.go b/pkg/util/http_test.go index 46d1e19..0073bfc 100644 --- a/pkg/util/http_test.go +++ b/pkg/util/http_test.go @@ -6,38 +6,38 @@ import ( "testing" "github.com/dgate-io/dgate/pkg/util" - "github.com/stretchr/testify/require" + "github.com/stretchr/testify/assert" ) func TestGetTrustedIP_Depth(t *testing.T) { req := requestWithXForwardedFor(t, "1.2.3.4", "1.2.3.5", "1.2.3.6") t.Run("Depth 0", func(t *testing.T) { - require.Equal(t, util.GetTrustedIP(req, 0), "127.0.0.1") + assert.Equal(t, util.GetTrustedIP(req, 0), "127.0.0.1") }) t.Run("Depth 1", func(t *testing.T) { - require.Equal(t, util.GetTrustedIP(req, 1), "1.2.3.6") + assert.Equal(t, util.GetTrustedIP(req, 1), "1.2.3.6") }) t.Run("Depth 2", func(t *testing.T) { - require.Equal(t, util.GetTrustedIP(req, 2), "1.2.3.5") + assert.Equal(t, util.GetTrustedIP(req, 2), "1.2.3.5") }) t.Run("Depth 3", func(t *testing.T) { - require.Equal(t, util.GetTrustedIP(req, 3), "1.2.3.4") + assert.Equal(t, util.GetTrustedIP(req, 3), "1.2.3.4") }) t.Run("Depth too High", func(t *testing.T) { - require.Equal(t, util.GetTrustedIP(req, 4), "1.2.3.4") - require.Equal(t, util.GetTrustedIP(req, 8), "1.2.3.4") - require.Equal(t, util.GetTrustedIP(req, 16), "1.2.3.4") + assert.Equal(t, util.GetTrustedIP(req, 4), "1.2.3.4") + assert.Equal(t, util.GetTrustedIP(req, 8), "1.2.3.4") + assert.Equal(t, util.GetTrustedIP(req, 16), "1.2.3.4") }) t.Run("Depth too Low", func(t *testing.T) { - require.Equal(t, util.GetTrustedIP(req, -1), "127.0.0.1") - require.Equal(t, util.GetTrustedIP(req, -10), "127.0.0.1") - require.Equal(t, util.GetTrustedIP(req, math.MinInt), "127.0.0.1") + assert.Equal(t, util.GetTrustedIP(req, -1), "127.0.0.1") + assert.Equal(t, util.GetTrustedIP(req, -10), "127.0.0.1") + assert.Equal(t, util.GetTrustedIP(req, math.MinInt), "127.0.0.1") }) } diff --git a/pkg/util/parse.go b/pkg/util/parse.go index 7787725..1e2c683 100644 --- a/pkg/util/parse.go +++ b/pkg/util/parse.go @@ -1,6 +1,9 @@ package util -import "strconv" +import ( + "strconv" + "time" +) func ParseInt(s string, def int) (int, error) { if s == "" { @@ -12,3 +15,12 @@ func ParseInt(s string, def int) (int, error) { return def, err } } + +func ParseBase36Timestamp(s string) (time.Time, error) { + if i, err := strconv.ParseInt(s, 36, 64); err == nil { + + return time.Unix(0, i), nil + } else { + return time.Time{}, err + } +} diff --git a/pkg/util/parse_test.go b/pkg/util/parse_test.go new file mode 100644 index 0000000..74cdf10 --- /dev/null +++ b/pkg/util/parse_test.go @@ -0,0 +1,20 @@ +package util_test + +import ( + "strconv" + "testing" + "time" + + "github.com/dgate-io/dgate/pkg/util" + "github.com/stretchr/testify/assert" +) + +func TestParseBase36Timestamp(t *testing.T) { + originalTime := time.Unix(0, time.Now().UnixNano()) + base36String := strconv.FormatInt(originalTime.UnixNano(), 36) + if parsedTime, err := util.ParseBase36Timestamp(base36String); err != nil { + t.Errorf("unexpected error: %v", err) + } else { + assert.Equal(t, parsedTime, originalTime) + } +}