Skip to content
This repository was archived by the owner on Nov 7, 2025. It is now read-only.

Commit d2cef94

Browse files
pdelewskipdelewski
authored
Adding cmd experimental, cleaning cmd main (#1352)
This PR: - cleans up main.go, removing commented or experimental code. - adds an experimental module with a few copied dependent files. <!-- A note on testing your PR --> <!-- Basic unit test run is executed against each commit in the PR. If you want to run a full integration test suite, you can trigger it by commenting with '/run-integration-tests' -->
1 parent 5e41c30 commit d2cef94

File tree

8 files changed

+1014
-77
lines changed

8 files changed

+1014
-77
lines changed

‎cmd/experimental/auth_middleware.go‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
// Copyright Quesma, licensed under the Elastic License 2.0.
2+
// SPDX-License-Identifier: Elastic-2.0
3+
package main
4+
5+
import (
6+
"github.com/QuesmaOrg/quesma/platform/config"
7+
"github.com/QuesmaOrg/quesma/platform/elasticsearch"
8+
"github.com/QuesmaOrg/quesma/platform/logger"
9+
"github.com/QuesmaOrg/quesma/platform/util"
10+
"net/http"
11+
"sync"
12+
"time"
13+
)
14+
15+
const cacheWipeInterval = 10 * time.Minute
16+
17+
// authMiddleware a simple implementation of an authentication middleware,
18+
// which checks the Authorization header and validates it against Elasticsearch.
19+
//
20+
// If the validation is positive, the Authorization header is stored in a cache to avoid unnecessary calls to Elasticsearch preceding each request.
21+
// The cache is wiped every 10 minutes - all items at once, perhaps this could be revisited in the future.
22+
type authMiddleware struct {
23+
nextHttpHandler http.Handler
24+
authHeaderCache sync.Map
25+
cacheWipeInterval time.Duration
26+
esClient elasticsearch.SimpleClient
27+
v2 bool
28+
}
29+
30+
func NewAuthMiddleware(next http.Handler, esConf config.ElasticsearchConfiguration) http.Handler {
31+
esClient := elasticsearch.NewSimpleClient(&esConf)
32+
middleware := &authMiddleware{nextHttpHandler: next, esClient: *esClient, cacheWipeInterval: cacheWipeInterval, v2: false}
33+
go middleware.startCacheWipeScheduler()
34+
return middleware
35+
}
36+
37+
func (a *authMiddleware) ServeHTTP(w http.ResponseWriter, r *http.Request) {
38+
auth := r.Header.Get("Authorization")
39+
if auth == "" {
40+
logger.Warn().Msgf("[AUTH] [%s] called without authorisation header, consider applying `disableAuth` option to the frontend connector to enable unauthorized access", r.URL)
41+
http.Error(w, "Unauthorized", http.StatusUnauthorized)
42+
return
43+
}
44+
45+
var userName string
46+
if user, err := util.ExtractUsernameFromBasicAuthHeader(auth); err == nil {
47+
userName = user
48+
} else {
49+
logger.Warn().Msgf("Failed to extract username from auth header: %v", err)
50+
}
51+
if _, ok := a.authHeaderCache.Load(auth); ok {
52+
logger.Debug().Msgf("[AUTH] [%s] called by [%s] - credentials loaded from cache", r.URL, userName)
53+
if !a.v2 {
54+
a.nextHttpHandler.ServeHTTP(w, r)
55+
}
56+
return
57+
}
58+
59+
if authenticated := a.esClient.Authenticate(r.Context(), auth); authenticated {
60+
logger.DebugWithCtx(r.Context()).Msgf("[AUTH] [%s] called by [%s] - authenticated against Elasticsearch, storing in cache", r.URL, userName)
61+
a.authHeaderCache.Store(auth, struct{}{})
62+
} else {
63+
logger.DebugWithCtx(r.Context()).Msgf("[AUTH] [%s] called by [%s] - authentication against Elasticsearch failed", r.URL, userName)
64+
http.Error(w, "Unauthorized", http.StatusUnauthorized)
65+
return
66+
}
67+
if !a.v2 {
68+
a.nextHttpHandler.ServeHTTP(w, r)
69+
}
70+
}
71+
72+
func (a *authMiddleware) startCacheWipeScheduler() {
73+
defer func() {
74+
if r := recover(); r != nil {
75+
logger.Error().Msgf("Recovered from panic during auth middleware cache wiping: [%v]", r)
76+
}
77+
}()
78+
ticker := time.NewTicker(a.cacheWipeInterval)
79+
defer ticker.Stop()
80+
for {
81+
<-ticker.C
82+
a.wipeCache()
83+
}
84+
}
85+
86+
func (a *authMiddleware) wipeCache() {
87+
logger.Debug().Msgf("[AUTH] wiping auth header cache")
88+
a.authHeaderCache.Range(func(key, value interface{}) bool {
89+
a.authHeaderCache.Delete(key)
90+
return true
91+
})
92+
}
93+
94+
func NewAuthMiddlewareV2(esConf config.ElasticsearchConfiguration) http.Handler {
95+
esClient := elasticsearch.NewSimpleClient(&esConf)
96+
middleware := &authMiddleware{esClient: *esClient, cacheWipeInterval: cacheWipeInterval, v2: true}
97+
go middleware.startCacheWipeScheduler()
98+
return middleware
99+
}

‎cmd/experimental/dual_write_proxy_v2.go‎

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
// Copyright Quesma, licensed under the Elastic License 2.0.
2+
// SPDX-License-Identifier: Elastic-2.0
3+
package main
4+
5+
import (
6+
"context"
7+
"github.com/QuesmaOrg/quesma/platform/ab_testing"
8+
"github.com/QuesmaOrg/quesma/platform/async_search_storage"
9+
"github.com/QuesmaOrg/quesma/platform/backend_connectors"
10+
"github.com/QuesmaOrg/quesma/platform/clickhouse"
11+
"github.com/QuesmaOrg/quesma/platform/config"
12+
"github.com/QuesmaOrg/quesma/platform/frontend_connectors"
13+
"github.com/QuesmaOrg/quesma/platform/ingest"
14+
"github.com/QuesmaOrg/quesma/platform/logger"
15+
"github.com/QuesmaOrg/quesma/platform/parsers/elastic_query_dsl"
16+
"github.com/QuesmaOrg/quesma/platform/schema"
17+
"github.com/QuesmaOrg/quesma/platform/table_resolver"
18+
"github.com/QuesmaOrg/quesma/platform/util"
19+
quesma_api "github.com/QuesmaOrg/quesma/platform/v2/core"
20+
"net/http"
21+
"strconv"
22+
"sync/atomic"
23+
)
24+
25+
const concurrentClientsLimitV2 = 100 // FIXME this should be configurable
26+
27+
type simultaneousClientsLimiterV2 struct {
28+
counter atomic.Int64
29+
limit int64
30+
}
31+
32+
func newSimultaneousClientsLimiterV2(limit int64) *simultaneousClientsLimiterV2 {
33+
return &simultaneousClientsLimiterV2{
34+
limit: limit,
35+
}
36+
}
37+
38+
func (c *simultaneousClientsLimiterV2) ServeHTTP(w http.ResponseWriter, r *http.Request) {
39+
40+
current := c.counter.Load()
41+
// this is hard limit, we should not allow to go over it
42+
if current >= c.limit {
43+
logger.ErrorWithCtx(r.Context()).Msgf("Too many requests. current: %d, limit: %d", current, c.limit)
44+
http.Error(w, "Too many requests", http.StatusTooManyRequests)
45+
return
46+
}
47+
48+
c.counter.Add(1)
49+
defer c.counter.Add(-1)
50+
}
51+
52+
type dualWriteHttpProxyV2 struct {
53+
quesmaV2 quesma_api.QuesmaBuilder
54+
logManager *clickhouse.LogManager
55+
publicPort util.Port
56+
asyncQueriesEvictor *async_search_storage.AsyncQueriesEvictor
57+
queryRunner *frontend_connectors.QueryRunner
58+
schemaRegistry schema.Registry
59+
schemaLoader clickhouse.TableDiscovery
60+
}
61+
62+
func (q *dualWriteHttpProxyV2) Stop(ctx context.Context) {
63+
q.Close(ctx)
64+
}
65+
66+
func newDualWriteProxyV2(dependencies quesma_api.Dependencies, schemaLoader clickhouse.TableDiscovery, logManager *clickhouse.LogManager, registry schema.Registry, config *config.QuesmaConfiguration, ingestProcessor *ingest.IngestProcessor, resolver table_resolver.TableResolver, abResultsRepository ab_testing.Sender) *dualWriteHttpProxyV2 {
67+
68+
queryProcessor := frontend_connectors.NewQueryRunner(logManager, config, dependencies.DebugInfoCollector(), registry, abResultsRepository, resolver, schemaLoader)
69+
70+
// not sure how we should configure our query translator ???
71+
// is this a config option??
72+
73+
queryProcessor.DateMathRenderer = elastic_query_dsl.DateMathExpressionFormatLiteral
74+
75+
// tests should not be run with optimization enabled by default
76+
queryProcessor.EnableQueryOptimization(config)
77+
esConn := backend_connectors.NewElasticsearchBackendConnector(config.Elasticsearch)
78+
79+
ingestRouter := frontend_connectors.ConfigureIngestRouterV2(config, dependencies, ingestProcessor, resolver, esConn)
80+
searchRouter := frontend_connectors.ConfigureSearchRouterV2(config, dependencies, registry, logManager, queryProcessor, resolver)
81+
82+
elasticHttpIngestFrontendConnector := frontend_connectors.NewElasticHttpIngestFrontendConnector(":"+strconv.Itoa(int(config.PublicTcpPort)),
83+
logManager, registry, config, ingestRouter)
84+
85+
elasticHttpQueryFrontendConnector := frontend_connectors.NewElasticHttpQueryFrontendConnector(":"+strconv.Itoa(int(config.PublicTcpPort)),
86+
logManager, registry, config, searchRouter)
87+
88+
quesmaBuilder := quesma_api.NewQuesma(dependencies)
89+
ingestPipeline := quesma_api.NewPipeline()
90+
ingestPipeline.AddFrontendConnector(elasticHttpIngestFrontendConnector)
91+
92+
queryPipeline := quesma_api.NewPipeline()
93+
queryPipeline.AddFrontendConnector(elasticHttpQueryFrontendConnector)
94+
quesmaBuilder.AddPipeline(queryPipeline)
95+
quesmaBuilder.AddPipeline(ingestPipeline)
96+
97+
quesmaV2, err := quesmaBuilder.Build()
98+
if err != nil {
99+
logger.Fatal().Msgf("Error building Quesma: %v", err)
100+
}
101+
if config.DisableAuth {
102+
elasticHttpIngestFrontendConnector.AddMiddleware(newSimultaneousClientsLimiterV2(concurrentClientsLimitV2))
103+
elasticHttpQueryFrontendConnector.AddMiddleware(newSimultaneousClientsLimiterV2(concurrentClientsLimitV2))
104+
} else {
105+
elasticHttpQueryFrontendConnector.AddMiddleware(newSimultaneousClientsLimiterV2(concurrentClientsLimitV2))
106+
elasticHttpQueryFrontendConnector.AddMiddleware(NewAuthMiddlewareV2(config.Elasticsearch))
107+
elasticHttpIngestFrontendConnector.AddMiddleware(newSimultaneousClientsLimiterV2(concurrentClientsLimitV2))
108+
elasticHttpIngestFrontendConnector.AddMiddleware(NewAuthMiddlewareV2(config.Elasticsearch))
109+
}
110+
111+
return &dualWriteHttpProxyV2{
112+
schemaRegistry: registry,
113+
schemaLoader: schemaLoader,
114+
quesmaV2: quesmaV2,
115+
logManager: logManager,
116+
publicPort: config.PublicTcpPort,
117+
asyncQueriesEvictor: async_search_storage.NewAsyncQueriesEvictor(
118+
queryProcessor.AsyncRequestStorage.(async_search_storage.AsyncSearchStorageInMemory),
119+
queryProcessor.AsyncQueriesContexts.(async_search_storage.AsyncQueryContextStorageInMemory),
120+
),
121+
queryRunner: queryProcessor,
122+
}
123+
}
124+
125+
func (q *dualWriteHttpProxyV2) Close(ctx context.Context) {
126+
if q.logManager != nil {
127+
defer q.logManager.Close()
128+
}
129+
if q.queryRunner != nil {
130+
q.queryRunner.Close()
131+
}
132+
if q.asyncQueriesEvictor != nil {
133+
q.asyncQueriesEvictor.Close()
134+
}
135+
q.quesmaV2.Stop(ctx)
136+
}
137+
138+
func (q *dualWriteHttpProxyV2) Ingest() {
139+
q.schemaLoader.ReloadTableDefinitions()
140+
q.logManager.Start()
141+
go q.asyncQueriesEvictor.AsyncQueriesGC()
142+
q.quesmaV2.Start()
143+
}

‎cmd/experimental/go.mod‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
module github.com/QuesmaOrg/quesma/quesma-cli-experimental
2+
3+
go 1.24.0
4+
5+
replace github.com/QuesmaOrg/quesma/platform => ../../platform
6+
7+
require github.com/QuesmaOrg/quesma/platform v0.0.0-00010101000000-000000000000
8+
9+
require (
10+
filippo.io/edwards25519 v1.1.0 // indirect
11+
github.com/ClickHouse/ch-go v0.65.1 // indirect
12+
github.com/ClickHouse/clickhouse-go/v2 v2.32.2 // indirect
13+
github.com/DATA-DOG/go-sqlmock v1.5.2 // indirect
14+
github.com/DataDog/go-sqllexer v0.0.21 // indirect
15+
github.com/H0llyW00dzZ/cidr v1.2.1 // indirect
16+
github.com/andybalholm/brotli v1.1.1 // indirect
17+
github.com/apparentlymart/go-cidr v1.1.0 // indirect
18+
github.com/barkimedes/go-deepcopy v0.0.0-20220514131651-17c30cfc62df // indirect
19+
github.com/beorn7/perks v1.0.1 // indirect
20+
github.com/cespare/xxhash/v2 v2.3.0 // indirect
21+
github.com/coreos/go-semver v0.3.1 // indirect
22+
github.com/fsnotify/fsnotify v1.8.0 // indirect
23+
github.com/go-faster/city v1.0.1 // indirect
24+
github.com/go-faster/errors v0.7.1 // indirect
25+
github.com/go-ole/go-ole v1.3.0 // indirect
26+
github.com/go-sql-driver/mysql v1.9.0 // indirect
27+
github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
28+
github.com/goccy/go-json v0.10.5 // indirect
29+
github.com/golang/glog v1.2.4 // indirect
30+
github.com/google/go-cmp v0.7.0 // indirect
31+
github.com/google/uuid v1.6.0 // indirect
32+
github.com/gorilla/mux v1.8.1 // indirect
33+
github.com/gorilla/securecookie v1.1.2 // indirect
34+
github.com/gorilla/sessions v1.4.0 // indirect
35+
github.com/hashicorp/errwrap v1.1.0 // indirect
36+
github.com/hashicorp/go-multierror v1.1.1 // indirect
37+
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
38+
github.com/jackc/pgpassfile v1.0.0 // indirect
39+
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
40+
github.com/jackc/pgx/v5 v5.7.2 // indirect
41+
github.com/jackc/puddle/v2 v2.2.2 // indirect
42+
github.com/k0kubun/pp v3.0.1+incompatible // indirect
43+
github.com/klauspost/compress v1.18.0 // indirect
44+
github.com/knadh/koanf/maps v0.1.1 // indirect
45+
github.com/knadh/koanf/parsers/json v0.1.0 // indirect
46+
github.com/knadh/koanf/parsers/yaml v0.1.0 // indirect
47+
github.com/knadh/koanf/providers/file v1.1.2 // indirect
48+
github.com/knadh/koanf/v2 v2.1.2 // indirect
49+
github.com/lufia/plan9stats v0.0.0-20240909124753-873cd0166683 // indirect
50+
github.com/markbates/goth v1.80.0 // indirect
51+
github.com/mattn/go-colorable v0.1.14 // indirect
52+
github.com/mattn/go-isatty v0.0.20 // indirect
53+
github.com/mitchellh/copystructure v1.2.0 // indirect
54+
github.com/mitchellh/reflectwalk v1.0.2 // indirect
55+
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
56+
github.com/paulmach/orb v0.11.1 // indirect
57+
github.com/pierrec/lz4/v4 v4.1.22 // indirect
58+
github.com/pkg/errors v0.9.1 // indirect
59+
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
60+
github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
61+
github.com/prometheus/client_golang v1.21.0 // indirect
62+
github.com/prometheus/client_model v0.6.1 // indirect
63+
github.com/prometheus/common v0.62.0 // indirect
64+
github.com/prometheus/procfs v0.15.1 // indirect
65+
github.com/rs/zerolog v1.33.0 // indirect
66+
github.com/segmentio/asm v1.2.0 // indirect
67+
github.com/shirou/gopsutil/v3 v3.24.5 // indirect
68+
github.com/shoenig/go-m1cpu v0.1.6 // indirect
69+
github.com/shopspring/decimal v1.4.0 // indirect
70+
github.com/spf13/pflag v1.0.5 // indirect
71+
github.com/tailscale/hujson v0.0.0-20241010212012-29efb4a0184b // indirect
72+
github.com/tidwall/gjson v1.18.0 // indirect
73+
github.com/tidwall/match v1.1.1 // indirect
74+
github.com/tidwall/pretty v1.2.1 // indirect
75+
github.com/tidwall/sjson v1.2.5 // indirect
76+
github.com/tklauser/go-sysconf v0.3.14 // indirect
77+
github.com/tklauser/numcpus v0.9.0 // indirect
78+
github.com/ucarion/urlpath v0.0.0-20200424170820-7ccc79b76bbb // indirect
79+
github.com/yusufpapurcu/wmi v1.2.4 // indirect
80+
go.opentelemetry.io/otel v1.34.0 // indirect
81+
go.opentelemetry.io/otel/trace v1.34.0 // indirect
82+
golang.org/x/crypto v0.33.0 // indirect
83+
golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8 // indirect
84+
golang.org/x/oauth2 v0.27.0 // indirect
85+
golang.org/x/sync v0.11.0 // indirect
86+
golang.org/x/sys v0.30.0 // indirect
87+
golang.org/x/text v0.22.0 // indirect
88+
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
89+
google.golang.org/grpc v1.66.2 // indirect
90+
google.golang.org/protobuf v1.36.1 // indirect
91+
gopkg.in/yaml.v3 v3.0.1 // indirect
92+
vitess.io/vitess v0.21.2 // indirect
93+
)

0 commit comments

Comments
 (0)