Using new http frontend connector API (#1059)

This PR:
- introduces `ElasticHttpFrontendConnector` and integrate with the
existing routing implementation.
- extracts `elasticFallback` function

Author: pdelewski

quesma/quesma/dual_write_proxy_v2.go

@@ -110,24 +110,14 @@ func newDualWriteProxyV2(schemaLoader clickhouse.TableDiscovery, logManager *cli
 		return routerInstance.failedRequests.Load()
 	})
 
-	handler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-		defer recovery.LogPanic()
-		reqBody, err := peekBodyV2(req)
-		if err != nil {
-			http.Error(w, "Error reading request body", http.StatusInternalServerError)
-			return
-		}
-
-		ua := req.Header.Get("User-Agent")
-		agent.UserAgentCounters().Add(ua, 1)
+	elasticHttpFrontentConnector := NewElasticHttpFrontendConnector(":"+strconv.Itoa(int(config.PublicTcpPort)),
+		&routerInstance, searchRouter, ingestRouter, logManager, agent)
 
-		routerInstance.reroute(req.Context(), w, req, reqBody, searchRouter, ingestRouter, logManager)
-	})
 	var limitedHandler http.Handler
 	if config.DisableAuth {
-		limitedHandler = newSimultaneousClientsLimiterV2(handler, concurrentClientsLimitV2)
+		limitedHandler = newSimultaneousClientsLimiterV2(elasticHttpFrontentConnector, concurrentClientsLimitV2)
 	} else {
-		limitedHandler = newSimultaneousClientsLimiterV2(NewAuthMiddleware(handler, config.Elasticsearch), concurrentClientsLimitV2)
+		limitedHandler = newSimultaneousClientsLimiterV2(NewAuthMiddleware(elasticHttpFrontentConnector, config.Elasticsearch), concurrentClientsLimitV2)
 	}
 
 	return &dualWriteHttpProxyV2{
@@ -286,6 +276,69 @@ func (*routerV2) closedIndexResponse(ctx context.Context, w http.ResponseWriter,
 
 }
 
+func (r *routerV2) elasticFallback(decision *table_resolver.Decision,
+	ctx context.Context, w http.ResponseWriter,
+	req *http.Request, reqBody []byte, logManager *clickhouse.LogManager) {
+
+	var sendToElastic bool
+
+	if decision != nil {
+
+		if decision.Err != nil {
+			w.Header().Set(quesmaSourceHeader, quesmaSourceClickhouse)
+			addProductAndContentHeaders(req.Header, w.Header())
+			r.errorResponseV2(ctx, decision.Err, w)
+			return
+		}
+
+		if decision.IsClosed {
+			w.Header().Set(quesmaSourceHeader, quesmaSourceClickhouse)
+			addProductAndContentHeaders(req.Header, w.Header())
+			r.closedIndexResponse(ctx, w, decision.IndexPattern)
+			return
+		}
+
+		if decision.IsEmpty {
+			w.Header().Set(quesmaSourceHeader, quesmaSourceClickhouse)
+			addProductAndContentHeaders(req.Header, w.Header())
+			w.WriteHeader(http.StatusNoContent)
+			w.Write(queryparser.EmptySearchResponse(ctx))
+			return
+		}
+
+		for _, connector := range decision.UseConnectors {
+			if _, ok := connector.(*table_resolver.ConnectorDecisionElastic); ok {
+				// this is desired elastic call
+				sendToElastic = true
+				break
+			}
+		}
+
+	} else {
+		// this is fallback case
+		// in case we don't support sth, we should send it to Elastic
+		sendToElastic = true
+	}
+
+	if sendToElastic {
+		feature.AnalyzeUnsupportedCalls(ctx, req.Method, req.URL.Path, req.Header.Get(opaqueIdHeaderKey), logManager.ResolveIndexPattern)
+
+		rawResponse := <-r.sendHttpRequestToElastic(ctx, req, reqBody, true)
+		response := rawResponse.response
+		if response != nil {
+			responseFromElasticV2(ctx, response, w)
+		} else {
+			w.Header().Set(quesmaSourceHeader, quesmaSourceElastic)
+			w.WriteHeader(500)
+			if rawResponse.error != nil {
+				_, _ = w.Write([]byte(rawResponse.error.Error()))
+			}
+		}
+	} else {
+		r.errorResponseV2(ctx, end_user_errors.ErrNoConnector.New(fmt.Errorf("no connector found")), w)
+	}
+}
+
 func (r *routerV2) reroute(ctx context.Context, w http.ResponseWriter, req *http.Request, reqBody []byte, searchRouter *mux.PathRouter, ingestRouter *mux.PathRouter, logManager *clickhouse.LogManager) {
 	defer recovery.LogAndHandlePanic(ctx, func(err error) {
 		w.WriteHeader(500)
@@ -352,64 +405,7 @@ func (r *routerV2) reroute(ctx context.Context, w http.ResponseWriter, req *http
 			r.errorResponseV2(ctx, err, w)
 		}
 	} else {
-
-		var sendToElastic bool
-
-		if decision != nil {
-
-			if decision.Err != nil {
-				w.Header().Set(quesmaSourceHeader, quesmaSourceClickhouse)
-				addProductAndContentHeaders(req.Header, w.Header())
-				r.errorResponseV2(ctx, decision.Err, w)
-				return
-			}
-
-			if decision.IsClosed {
-				w.Header().Set(quesmaSourceHeader, quesmaSourceClickhouse)
-				addProductAndContentHeaders(req.Header, w.Header())
-				r.closedIndexResponse(ctx, w, decision.IndexPattern)
-				return
-			}
-
-			if decision.IsEmpty {
-				w.Header().Set(quesmaSourceHeader, quesmaSourceClickhouse)
-				addProductAndContentHeaders(req.Header, w.Header())
-				w.WriteHeader(http.StatusNoContent)
-				w.Write(queryparser.EmptySearchResponse(ctx))
-				return
-			}
-
-			for _, connector := range decision.UseConnectors {
-				if _, ok := connector.(*table_resolver.ConnectorDecisionElastic); ok {
-					// this is desired elastic call
-					sendToElastic = true
-					break
-				}
-			}
-
-		} else {
-			// this is fallback case
-			// in case we don't support sth, we should send it to Elastic
-			sendToElastic = true
-		}
-
-		if sendToElastic {
-			feature.AnalyzeUnsupportedCalls(ctx, req.Method, req.URL.Path, req.Header.Get(opaqueIdHeaderKey), logManager.ResolveIndexPattern)
-
-			rawResponse := <-r.sendHttpRequestToElastic(ctx, req, reqBody, true)
-			response := rawResponse.response
-			if response != nil {
-				responseFromElasticV2(ctx, response, w)
-			} else {
-				w.Header().Set(quesmaSourceHeader, quesmaSourceElastic)
-				w.WriteHeader(500)
-				if rawResponse.error != nil {
-					_, _ = w.Write([]byte(rawResponse.error.Error()))
-				}
-			}
-		} else {
-			r.errorResponseV2(ctx, end_user_errors.ErrNoConnector.New(fmt.Errorf("no connector found")), w)
-		}
+		r.elasticFallback(decision, ctx, w, req, reqBody, logManager)
 	}
 }
 

quesma/quesma/elastic_http_frontend_connector.go

@@ -0,0 +1,52 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+
+package quesma
+
+import (
+	"net/http"
+	"quesma/clickhouse"
+	"quesma/frontend_connectors"
+	"quesma/quesma/mux"
+	"quesma/quesma/recovery"
+	"quesma/telemetry"
+)
+
+type ElasticHttpFrontendConnector struct {
+	*frontend_connectors.BasicHTTPFrontendConnector
+	routerInstance *routerV2
+	searchRouter   *mux.PathRouter
+	ingestRouter   *mux.PathRouter
+	logManager     *clickhouse.LogManager
+	agent          telemetry.PhoneHomeAgent
+}
+
+func NewElasticHttpFrontendConnector(endpoint string,
+	routerInstance *routerV2,
+	searchRouter *mux.PathRouter,
+	ingestRouter *mux.PathRouter,
+	logManager *clickhouse.LogManager,
+	agent telemetry.PhoneHomeAgent) *ElasticHttpFrontendConnector {
+	return &ElasticHttpFrontendConnector{
+		BasicHTTPFrontendConnector: frontend_connectors.NewBasicHTTPFrontendConnector(endpoint),
+		routerInstance:             routerInstance,
+		searchRouter:               searchRouter,
+		ingestRouter:               ingestRouter,
+		logManager:                 logManager,
+		agent:                      agent,
+	}
+}
+
+func (h *ElasticHttpFrontendConnector) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	defer recovery.LogPanic()
+	reqBody, err := peekBodyV2(req)
+	if err != nil {
+		http.Error(w, "Error reading request body", http.StatusInternalServerError)
+		return
+	}
+
+	ua := req.Header.Get("User-Agent")
+	h.agent.UserAgentCounters().Add(ua, 1)
+
+	h.routerInstance.reroute(req.Context(), w, req, reqBody, h.searchRouter, h.ingestRouter, h.logManager)
+}