Fix _bulk endpoint (#1243)

Given that we have the sophisticated bulk body introduced in
#1202, there is no longer need
to examine the bulk payload in order to see where it should be routed.

Also, that `matchedAgainstBulkBody` implementation wasn't entirely
correct anyways (relying on optional data line being always %2).

Author: mieciu

quesma/quesma/es_responses.go

@@ -14,7 +14,6 @@ import (
 	quesma_api "github.com/QuesmaOrg/quesma/quesma/v2/core"
 	"github.com/goccy/go-json"
 	"net/http"
-	"regexp"
 	"sync"
 )
 
@@ -212,15 +211,3 @@ type (
 		Index              string `json:"index"`
 	}
 )
-
-var indexNamePattern = regexp.MustCompile(`"_index"\s*:\s*"([^"]+)"`)
-
-func extractIndexName(input string) string {
-	results := indexNamePattern.FindStringSubmatch(input)
-
-	if len(results) < 2 {
-		return ""
-	}
-
-	return results[1]
-}

quesma/quesma/matchers.go

@@ -5,7 +5,6 @@ package quesma
 import (
 	"github.com/QuesmaOrg/quesma/quesma/logger"
 	"github.com/QuesmaOrg/quesma/quesma/painful"
-	"github.com/QuesmaOrg/quesma/quesma/quesma/config"
 	"github.com/QuesmaOrg/quesma/quesma/quesma/types"
 	"github.com/QuesmaOrg/quesma/quesma/table_resolver"
 	"github.com/QuesmaOrg/quesma/quesma/v2/core"
@@ -24,38 +23,6 @@ func matchedAgainstAsyncId() quesma_api.RequestMatcher {
 	})
 }
 
-func matchedAgainstBulkBody(configuration *config.QuesmaConfiguration, tableResolver table_resolver.TableResolver) quesma_api.RequestMatcher {
-	return quesma_api.RequestMatcherFunc(func(req *quesma_api.Request) quesma_api.MatchResult {
-		idx := 0
-		for _, s := range strings.Split(req.Body, "\n") {
-			if len(s) == 0 {
-				// ElasticSearch Agent sends empty lines between some JSONs, ignore them.
-				continue
-			}
-			if idx%2 == 0 {
-				name := extractIndexName(s)
-
-				decision := tableResolver.Resolve(quesma_api.IngestPipeline, name)
-
-				if decision.IsClosed {
-					return quesma_api.MatchResult{Matched: true, Decision: decision}
-				}
-
-				// if have any enabled Clickhouse connector, then return true
-				for _, connector := range decision.UseConnectors {
-					if _, ok := connector.(*quesma_api.ConnectorDecisionClickhouse); ok {
-						return quesma_api.MatchResult{Matched: true, Decision: decision}
-					}
-				}
-			}
-			idx += 1
-		}
-
-		// All indexes are disabled, the whole bulk can go to Elastic
-		return quesma_api.MatchResult{Matched: false}
-	})
-}
-
 // Query path only (looks at QueryTarget)
 func matchedAgainstPattern(indexRegistry table_resolver.TableResolver) quesma_api.RequestMatcher {
 	return matchAgainstTableResolver(indexRegistry, quesma_api.QueryPipeline)

quesma/quesma/router_test.go

@@ -98,7 +98,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		return elasticsearchQueryResult(`{"cluster_name": "quesma"}`, http.StatusOK), nil
 	})
 
-	router.Register(routes.BulkPath, and(method("POST", "PUT"), matchedAgainstBulkBody(cfg, tableResolver)), func(ctx context.Context, req *quesma_api.Request, _ http.ResponseWriter) (*quesma_api.Result, error) {
+	router.Register(routes.BulkPath, method("POST", "PUT"), func(ctx context.Context, req *quesma_api.Request, _ http.ResponseWriter) (*quesma_api.Result, error) {
 
 		body, err := types.ExpectNDJSON(req.ParsedBody)
 		if err != nil {
@@ -682,60 +682,6 @@ func withAutodiscovery(cfg config.QuesmaConfiguration) config.QuesmaConfiguratio
 	return cfg
 }
 
-func Test_matchedAgainstBulkBody(t *testing.T) {
-
-	t.Skip(skipMessage)
-
-	tests := []struct {
-		name   string
-		body   string
-		config config.QuesmaConfiguration
-		want   bool
-	}{
-		{
-			name:   "single index, config present",
-			body:   `{"create":{"_index":"logs-generic-default"}}`,
-			config: indexConfig("logs-generic-default", false),
-			want:   true,
-		},
-		{
-			name:   "single index, table not present",
-			body:   `{"create":{"_index":"logs-generic-default"}}`,
-			config: indexConfig("foo", false),
-			want:   false,
-		},
-		{
-			name:   "multiple indexes, table present",
-			body:   `{"create":{"_index":"logs-generic-default"}}` + "\n{}\n" + `{"create":{"_index":"logs-generic-default"}}`,
-			config: indexConfig("logs-generic-default", false),
-			want:   true,
-		},
-		{
-			name:   "multiple indexes, some tables not present",
-			body:   `{"create":{"_index":"logs-generic-default"}}` + "\n{}\n" + `{"create":{"_index":"non-existent"}}`,
-			config: indexConfig("logs-generic-default", false),
-			want:   true,
-		},
-		{
-			name:   "multiple indexes, all tables not present",
-			body:   `{"create":{"_index":"not-there"}}` + "\n{}\n" + `{"create":{"_index":"non-existent"}}`,
-			config: indexConfig("logs-generic-default", false),
-			want:   false,
-		},
-	}
-
-	resolver := table_resolver.NewEmptyTableResolver()
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-
-			req := &quesma_api.Request{Body: tt.body}
-
-			assert.Equalf(t, tt.want, matchedAgainstBulkBody(&tt.config, resolver).Matches(req), "matchedAgainstBulkBody(%+v)", tt.config)
-		})
-	}
-}
-
 const testIndexName = "indexName"
 
 func TestConfigureRouter(t *testing.T) {

quesma/quesma/router_v2.go

@@ -68,7 +68,7 @@ func ConfigureIngestRouterV2(cfg *config.QuesmaConfiguration, dependencies quesm
 		}, nil
 	})
 
-	router.Register(routes.BulkPath, and(method("POST", "PUT"), matchedAgainstBulkBody(cfg, tableResolver)), func(ctx context.Context, req *quesma_api.Request, _ http.ResponseWriter) (*quesma_api.Result, error) {
+	router.Register(routes.BulkPath, method("POST", "PUT"), func(ctx context.Context, req *quesma_api.Request, _ http.ResponseWriter) (*quesma_api.Result, error) {
 		body, err := types.ExpectNDJSON(req.ParsedBody)
 		if err != nil {
 			return nil, err