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

Commit 72cab08

Browse files
nablaonenablaone
authored
A/B testing - E vs C and C vs E (#919)
This PR contains: - generalized code of A/B testing query results for both scenarios: Elastic vs Clickhouse and Clickhouse vs Elastic - kibana dashboard id is stored in a separate column - we don't store requests and A/B responses if the responses match Some code is copy pasted. I'm not brave enough to generalize it now. Tests will be added as part of IT suite. <img width="789" alt="Screenshot 2024-10-28 at 12 37 16" src="https://github.com/user-attachments/assets/3be97a39-c245-4477-934c-40bf1a918947"> <img width="1995" alt="Screenshot 2024-10-28 at 12 36 49" src="https://github.com/user-attachments/assets/77c487f1-c232-4216-9f9d-779925cf8c25">
1 parent 8ccf821 commit 72cab08

File tree

12 files changed

+478
-259
lines changed

12 files changed

+478
-259
lines changed

‎quesma/ab_testing/collector/collector.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,12 +83,13 @@ func NewCollector(ctx context.Context, ingester ingest.Ingester, healthQueue cha
8383
&diffTransformer{},
8484
//&ppPrintFanout{},
8585
//&mismatchedOnlyFilter{},
86+
&redactOkResults{},
8687
//&elasticSearchFanout{
8788
// url: "http://localhost:8080",
8889
// indexName: "ab_testing_logs",
8990
//},
9091
&internalIngestFanout{
91-
indexName: "ab_testing_logs",
92+
indexName: ab_testing.ABTestingTableName,
9293
ingestProcessor: ingester,
9394
},
9495
},

‎quesma/ab_testing/collector/diff.go‎

Lines changed: 16 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -66,22 +66,31 @@ func (t *diffTransformer) process(in EnrichedResults) (out EnrichedResults, drop
6666
}
6767

6868
if len(mismatches) > 0 {
69-
b, err := json.MarshalIndent(mismatches, "", " ")
70-
71-
if err != nil {
72-
return in, false, fmt.Errorf("failed to marshal mismatches: %w", err)
73-
}
7469

75-
in.Mismatch.Mismatches = string(b)
7670
in.Mismatch.IsOK = false
7771
in.Mismatch.Count = len(mismatches)
78-
in.Mismatch.Message = mismatches.String()
7972

8073
topMismatchType, _ := t.mostCommonMismatchType(mismatches)
8174
if topMismatchType != "" {
8275
in.Mismatch.TopMismatchType = topMismatchType
8376
}
8477

78+
// if there are too many mismatches, we only show the first 20
79+
// this is to avoid overwhelming the user with too much information
80+
const mismatchesSize = 20
81+
82+
if len(mismatches) > mismatchesSize {
83+
mismatches = mismatches[:mismatchesSize]
84+
}
85+
86+
b, err := json.MarshalIndent(mismatches, "", " ")
87+
88+
if err != nil {
89+
return in, false, fmt.Errorf("failed to marshal mismatches: %w", err)
90+
}
91+
in.Mismatch.Mismatches = string(b)
92+
in.Mismatch.Message = mismatches.String()
93+
8594
} else {
8695
in.Mismatch.Mismatches = "[]"
8796
in.Mismatch.IsOK = true

‎quesma/ab_testing/collector/filters.go‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,3 +40,27 @@ func (t *mismatchedOnlyFilter) process(in EnrichedResults) (out EnrichedResults,
4040

4141
// avoid unused struct error
4242
var _ = &mismatchedOnlyFilter{}
43+
44+
type redactOkResults struct {
45+
}
46+
47+
func (t *redactOkResults) name() string {
48+
return "redactOkResults"
49+
}
50+
51+
func (t *redactOkResults) process(in EnrichedResults) (out EnrichedResults, drop bool, err error) {
52+
53+
// we're not interested in the details of the request and responses if the mismatch is OK
54+
55+
redactMsg := "***REDACTED***"
56+
if in.Mismatch.IsOK {
57+
in.Request.Body = redactMsg
58+
in.A.Body = redactMsg
59+
in.B.Body = redactMsg
60+
in.Mismatch.Message = "OK"
61+
}
62+
63+
return in, false, nil
64+
}
65+
66+
var _ = &redactOkResults{}

‎quesma/ab_testing/model.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22
// SPDX-License-Identifier: Elastic-2.0
33
package ab_testing
44

5+
const ABTestingTableName = "ab_testing_logs"
6+
57
type Request struct {
68
Path string `json:"path"`
79
IndexName string `json:"index_name"`

‎quesma/jsondiff/elastic_response_diff.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ import "fmt"
66

77
// NewElasticResponseJSONDiff creates a JSONDiff instance that is tailored to compare Elasticsearch response JSONs.
88
func NewElasticResponseJSONDiff() (*JSONDiff, error) {
9-
d, err := NewJSONDiff("^id$", ".*Quesma_key_.*", "^took$")
9+
d, err := NewJSONDiff("^id$", ".*Quesma_key_.*", "^took$", ".*__quesma_total_count", ".*\\._id", "^_shards.*", ".*\\._score", ".*\\._source", ".*\\.__quesma_originalKey")
1010

1111
if err != nil {
1212
return nil, fmt.Errorf("could not create JSONDiff: %v", err)

‎quesma/jsondiff/jsondiff.go‎

Lines changed: 37 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ func newType(code, message string) mismatchType {
2525
var (
2626
invalidType = newType("invalid_type", "Types are not equal")
2727
invalidValue = newType("invalid_value", "Values are not equal")
28+
invalidNumberValue = newType("invalid_number_value", "Numbers are not equal")
29+
invalidDateValue = newType("invalid_date_value", "Dates are not equal")
2830
invalidArrayLength = newType("invalid_array_length", "Array lengths are not equal")
2931
invalidArrayLengthOffByOne = newType("invalid_array_length_off_by_one", "Array lengths are off by one.")
3032
objectDifference = newType("object_difference", "Objects are different")
@@ -355,6 +357,8 @@ func (d *JSONDiff) asType(a any) string {
355357
return fmt.Sprintf("%T", a)
356358
}
357359

360+
var dateRx = regexp.MustCompile(`\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}`)
361+
358362
func (d *JSONDiff) compare(expected any, actual any) {
359363

360364
if d.isIgnoredPath() {
@@ -399,9 +403,9 @@ func (d *JSONDiff) compare(expected any, actual any) {
399403
case float64:
400404

401405
// float operations are noisy, we need to compare them with desired precision
402-
403-
epsilon := 1e-9
404-
relativeTolerance := 1e-9
406+
// this is lousy, but it works for now
407+
epsilon := 1e-3
408+
relativeTolerance := 1e-3
405409
aFloat := expected.(float64)
406410
bFloat := actual.(float64)
407411

@@ -411,8 +415,37 @@ func (d *JSONDiff) compare(expected any, actual any) {
411415
relativeDiff := absDiff / math.Max(math.Abs(aFloat), math.Abs(bFloat))
412416

413417
if relativeDiff > relativeTolerance {
414-
d.addMismatch(invalidValue, d.asValue(expected), d.asValue(actual))
418+
d.addMismatch(invalidNumberValue, d.asValue(expected), d.asValue(actual))
419+
}
420+
}
421+
422+
default:
423+
d.addMismatch(invalidType, d.asType(expected), d.asType(actual))
424+
}
425+
426+
case string:
427+
428+
switch actualString := actual.(type) {
429+
case string:
430+
431+
if dateRx.MatchString(aVal) && dateRx.MatchString(actualString) {
432+
433+
// TODO add better date comparison here
434+
// parse both date and compare them with desired precision
435+
436+
// elastics returns date in formats
437+
// "2024-10-24T00:00:00.000+02:00"
438+
// "2024-10-24T00:00:00.000Z"
439+
440+
// quesma returns
441+
// 2024-10-23T22:00:00.000
442+
compareOnly := "2000-01-"
443+
444+
if aVal[:len(compareOnly)] != actualString[:len(compareOnly)] {
445+
d.addMismatch(invalidDateValue, d.asValue(expected), d.asValue(actual))
415446
}
447+
448+
return
416449
}
417450

418451
default:

‎quesma/jsondiff/jsondiff_test.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ func TestJSONDiff(t *testing.T) {
3939
name: "Test 2",
4040
expected: `{"a": 1, "b": 2, "c": 3}`,
4141
actual: `{"a": 1, "b": 3, "c": 3}`,
42-
problems: []JSONMismatch{mismatch("b", invalidValue)},
42+
problems: []JSONMismatch{mismatch("b", invalidNumberValue)},
4343
},
4444

4545
{
@@ -67,7 +67,7 @@ func TestJSONDiff(t *testing.T) {
6767
name: "array element difference",
6868
expected: `{"a": [1, 2, 3], "b": 2, "c": 3}`,
6969
actual: `{"a": [1, 2, 4], "b": 2, "c": 3}`,
70-
problems: []JSONMismatch{mismatch("a.[2]", invalidValue)},
70+
problems: []JSONMismatch{mismatch("a.[2]", invalidNumberValue)},
7171
},
7272

7373
{
@@ -81,28 +81,28 @@ func TestJSONDiff(t *testing.T) {
8181
name: "object difference",
8282
expected: `{"a": {"b": 1}, "c": 3}`,
8383
actual: `{"a": {"b": 2}, "c": 3}`,
84-
problems: []JSONMismatch{mismatch("a.b", invalidValue)},
84+
problems: []JSONMismatch{mismatch("a.b", invalidNumberValue)},
8585
},
8686

8787
{
8888
name: "deep path difference",
8989
expected: `{"a": {"d": {"b": 1}}, "c": 3}`,
9090
actual: `{"a": {"d": {"b": 2}}, "c": 3}`,
91-
problems: []JSONMismatch{mismatch("a.d.b", invalidValue)},
91+
problems: []JSONMismatch{mismatch("a.d.b", invalidNumberValue)},
9292
},
9393

9494
{
9595
name: "deep path difference",
9696
expected: `{"a": {"d": {"b": 1}}, "c": 3, "_ignore": 1}`,
9797
actual: `{"a": {"d": {"b": 2}}, "c": 3}`,
98-
problems: []JSONMismatch{mismatch("a.d.b", invalidValue)},
98+
problems: []JSONMismatch{mismatch("a.d.b", invalidNumberValue)},
9999
},
100100

101101
{
102102
name: "array sort difference ",
103103
expected: `{"a": [1, 2, 3], "b": 2, "c": 3}`,
104104
actual: `{"a": [1, 3, 2], "b": 2, "c": 3}`,
105-
problems: []JSONMismatch{mismatch("a.[1]", invalidValue), mismatch("a.[2]", invalidValue)},
105+
problems: []JSONMismatch{mismatch("a.[1]", invalidNumberValue), mismatch("a.[2]", invalidNumberValue)},
106106
},
107107

108108
{

‎quesma/quesma/config/config_v2.go‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -578,8 +578,9 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
578578
}
579579
}
580580

581-
if len(processedConfig.QueryTarget) == 2 && !(processedConfig.QueryTarget[0] == ClickhouseTarget && processedConfig.QueryTarget[1] == ElasticsearchTarget) {
582-
errAcc = multierror.Append(errAcc, fmt.Errorf("index %s has invalid dual query target configuration - when you specify two targets, ClickHouse has to be the primary one and Elastic has to be the secondary one", indexName))
581+
if len(processedConfig.QueryTarget) == 2 && !((processedConfig.QueryTarget[0] == ClickhouseTarget && processedConfig.QueryTarget[1] == ElasticsearchTarget) ||
582+
(processedConfig.QueryTarget[0] == ElasticsearchTarget && processedConfig.QueryTarget[1] == ClickhouseTarget)) {
583+
errAcc = multierror.Append(errAcc, fmt.Errorf("index %s has invalid dual query target configuration", indexName))
583584
continue
584585
}
585586

@@ -676,10 +677,12 @@ func (c *QuesmaNewConfiguration) TranslateToLegacyConfig() QuesmaConfiguration {
676677
}
677678
}
678679

679-
if len(processedConfig.QueryTarget) == 2 && !(processedConfig.QueryTarget[0] == ClickhouseTarget && processedConfig.QueryTarget[1] == ElasticsearchTarget) {
680-
errAcc = multierror.Append(errAcc, fmt.Errorf("index %s has invalid dual query target configuration - when you specify two targets, ClickHouse has to be the primary one and Elastic has to be the secondary one", indexName))
680+
if len(processedConfig.QueryTarget) == 2 && !((processedConfig.QueryTarget[0] == ClickhouseTarget && processedConfig.QueryTarget[1] == ElasticsearchTarget) ||
681+
(processedConfig.QueryTarget[0] == ElasticsearchTarget && processedConfig.QueryTarget[1] == ClickhouseTarget)) {
682+
errAcc = multierror.Append(errAcc, fmt.Errorf("index %s has invalid dual query target configuration", indexName))
681683
continue
682684
}
685+
683686
if len(processedConfig.QueryTarget) == 2 {
684687
// Turn on A/B testing
685688
processedConfig.Optimizers = make(map[string]OptimizerConfiguration)

‎quesma/quesma/search.go‎

Lines changed: 12 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -203,7 +203,7 @@ func (q *QueryRunner) runExecutePlanAsync(ctx context.Context, plan *model.Execu
203203
}()
204204
}
205205

206-
func (q *QueryRunner) executePlan(ctx context.Context, plan *model.ExecutionPlan, queryTranslator IQueryTranslator, table *clickhouse.Table, body types.JSON, optAsync *AsyncQuery, optComparePlansCh chan<- executionPlanResult) (responseBody []byte, err error) {
206+
func (q *QueryRunner) executePlan(ctx context.Context, plan *model.ExecutionPlan, queryTranslator IQueryTranslator, table *clickhouse.Table, body types.JSON, optAsync *AsyncQuery, optComparePlansCh chan<- executionPlanResult, abTestingMainPlan bool) (responseBody []byte, err error) {
207207
contextValues := tracing.ExtractValues(ctx)
208208
id := contextValues.RequestId
209209
path := contextValues.RequestPath
@@ -214,7 +214,7 @@ func (q *QueryRunner) executePlan(ctx context.Context, plan *model.ExecutionPlan
214214
sendMainPlanResult := func(responseBody []byte, err error) {
215215
if optComparePlansCh != nil {
216216
optComparePlansCh <- executionPlanResult{
217-
isMain: true,
217+
isMain: abTestingMainPlan,
218218
plan: plan,
219219
err: err,
220220
responseBody: responseBody,
@@ -300,31 +300,27 @@ func (q *QueryRunner) handleSearchCommon(ctx context.Context, indexPattern strin
300300
return nil, end_user_errors.ErrSearchCondition.New(fmt.Errorf("no connectors to use"))
301301
}
302302

303-
var clickhouseDecision *table_resolver.ConnectorDecisionClickhouse
304-
var elasticDecision *table_resolver.ConnectorDecisionElastic
303+
var clickhouseConnector *table_resolver.ConnectorDecisionClickhouse
304+
305305
for _, connector := range decision.UseConnectors {
306306
switch c := connector.(type) {
307307

308308
case *table_resolver.ConnectorDecisionClickhouse:
309-
clickhouseDecision = c
309+
clickhouseConnector = c
310310

311311
case *table_resolver.ConnectorDecisionElastic:
312-
elasticDecision = c
312+
// NOP
313313

314314
default:
315315
return nil, fmt.Errorf("unknown connector type: %T", c)
316316
}
317317
}
318318

319319
// it's impossible here to don't have a clickhouse decision
320-
if clickhouseDecision == nil {
320+
if clickhouseConnector == nil {
321321
return nil, fmt.Errorf("no clickhouse connector")
322322
}
323323

324-
if elasticDecision != nil {
325-
fmt.Println("elastic", elasticDecision)
326-
}
327-
328324
var responseBody []byte
329325

330326
startTime := time.Now()
@@ -343,7 +339,7 @@ func (q *QueryRunner) handleSearchCommon(ctx context.Context, indexPattern strin
343339

344340
var table *clickhouse.Table // TODO we should use schema here only
345341
var currentSchema schema.Schema
346-
resolvedIndexes := clickhouseDecision.ClickhouseTables
342+
resolvedIndexes := clickhouseConnector.ClickhouseTables
347343

348344
if len(resolvedIndexes) == 1 {
349345
indexName := resolvedIndexes[0] // we got exactly one table here because of the check above
@@ -446,17 +442,12 @@ func (q *QueryRunner) handleSearchCommon(ctx context.Context, indexPattern strin
446442
plan.StartTime = startTime
447443
plan.Name = model.MainExecutionPlan
448444

449-
// Some flags may trigger alternative execution plans, this is primary for dev
450-
451-
alternativePlan, alternativePlanExecutor := q.maybeCreateAlternativeExecutionPlan(ctx, resolvedIndexes, plan, queryTranslator, body, table, optAsync != nil)
452-
453-
var optComparePlansCh chan<- executionPlanResult
454-
455-
if alternativePlan != nil {
456-
optComparePlansCh = q.runAlternativePlanAndComparison(ctx, alternativePlan, alternativePlanExecutor, body)
445+
if decision.EnableABTesting {
446+
return q.executeABTesting(ctx, plan, queryTranslator, table, body, optAsync, decision, indexPattern)
457447
}
458448

459-
return q.executePlan(ctx, plan, queryTranslator, table, body, optAsync, optComparePlansCh)
449+
return q.executePlan(ctx, plan, queryTranslator, table, body, optAsync, nil, true)
450+
460451
}
461452

462453
func (q *QueryRunner) storeAsyncSearch(qmc *ui.QuesmaManagementConsole, id, asyncId string,

0 commit comments

Comments
 (0)