Separate query-service functionality from http handler

cmd/query/app/handler_archive_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 
+	"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"
 	"github.com/jaegertracing/jaeger/model"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 	spanstoremocks "github.com/jaegertracing/jaeger/storage/spanstore/mocks"
@@ -42,7 +43,7 @@ func TestGetArchivedTrace_NotFound(t *testing.T) {
 				assert.EqualError(t, err,
 					`404 error from server: {"data":null,"total":0,"limit":0,"offset":0,"errors":[{"code":404,"msg":"trace not found"}]}`+"\n",
 				)
-			}, HandlerOptions.ArchiveSpanReader(archiveReader)) // nil is ok
+			}, querysvc.QueryServiceOptions{ArchiveSpanReader: archiveReader}) // nil is ok
 		})
 	}
 }
@@ -62,15 +63,15 @@ func TestGetArchivedTraceSuccess(t *testing.T) {
 		assert.Len(t, response.Errors, 0)
 		assert.Len(t, response.Traces, 1)
 		assert.Equal(t, traceID.String(), string(response.Traces[0].TraceID))
-	}, HandlerOptions.ArchiveSpanReader(mockReader))
+	}, querysvc.QueryServiceOptions{ArchiveSpanReader: mockReader})
 }
 
 func TestArchiveTrace_NoStorage(t *testing.T) {
 	withTestServer(t, func(ts *testServer) {
 		var response structuredResponse
 		err := postJSON(ts.server.URL+"/api/archive/"+mockTraceID.String(), []string{}, &response)
 		assert.EqualError(t, err, `500 error from server: {"data":null,"total":0,"limit":0,"offset":0,"errors":[{"code":500,"msg":"archive span storage was not configured"}]}`+"\n")
-	})
+	}, querysvc.QueryServiceOptions{})
 }
 
 func TestArchiveTrace_Success(t *testing.T) {
@@ -83,7 +84,7 @@ func TestArchiveTrace_Success(t *testing.T) {
 		var response structuredResponse
 		err := postJSON(ts.server.URL+"/api/archive/"+mockTraceID.String(), []string{}, &response)
 		assert.NoError(t, err)
-	}, HandlerOptions.ArchiveSpanWriter(mockWriter))
+	}, querysvc.QueryServiceOptions{ArchiveSpanWriter: mockWriter})
 }
 
 func TestArchiveTrace_WriteErrors(t *testing.T) {
@@ -96,5 +97,5 @@ func TestArchiveTrace_WriteErrors(t *testing.T) {
 		var response structuredResponse
 		err := postJSON(ts.server.URL+"/api/archive/"+mockTraceID.String(), []string{}, &response)
 		assert.EqualError(t, err, `500 error from server: {"data":null,"total":0,"limit":0,"offset":0,"errors":[{"code":500,"msg":"[cannot save, cannot save]"}]}`+"\n")
-	}, HandlerOptions.ArchiveSpanWriter(mockWriter))
+	}, querysvc.QueryServiceOptions{ArchiveSpanWriter: mockWriter})
 }

cmd/query/app/handler_options.go

@@ -19,9 +19,6 @@ import (
 
 	"github.com/opentracing/opentracing-go"
 	"go.uber.org/zap"
-
-	"github.com/jaegertracing/jaeger/model/adjuster"
-	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
 
 // HandlerOption is a function that sets some option on the APIHandler
@@ -40,13 +37,6 @@ func (handlerOptions) Logger(logger *zap.Logger) HandlerOption {
 	}
 }
 
-// Adjusters creates a HandlerOption that initializes the sequence of Adjusters on the APIHandler,
-func (handlerOptions) Adjusters(adjusters ...adjuster.Adjuster) HandlerOption {
-	return func(apiHandler *APIHandler) {
-		apiHandler.adjuster = adjuster.Sequence(adjusters...)
-	}
-}
-
 // BasePath creates a HandlerOption that initializes the base path for all HTTP routes
 func (handlerOptions) BasePath(prefix string) HandlerOption {
 	return func(apiHandler *APIHandler) {
@@ -68,20 +58,6 @@ func (handlerOptions) QueryLookbackDuration(queryLookbackDuration time.Duration)
 	}
 }
 
-// ArchiveSpanReader creates a HandlerOption that initializes lookback duration
-func (handlerOptions) ArchiveSpanReader(reader spanstore.Reader) HandlerOption {
-	return func(apiHandler *APIHandler) {
-		apiHandler.archiveSpanReader = reader
-	}
-}
-
-// ArchiveSpanWriter creates a HandlerOption that initializes lookback duration
-func (handlerOptions) ArchiveSpanWriter(writer spanstore.Writer) HandlerOption {
-	return func(apiHandler *APIHandler) {
-		apiHandler.archiveSpanWriter = writer
-	}
-}
-
 // Tracer creates a HandlerOption that initializes OpenTracing tracer
 func (handlerOptions) Tracer(tracer opentracing.Tracer) HandlerOption {
 	return func(apiHandler *APIHandler) {

cmd/query/app/handler.go → cmd/query/app/http_handler.go

@@ -29,12 +29,11 @@ import (
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 
+	"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"
 	"github.com/jaegertracing/jaeger/model"
-	"github.com/jaegertracing/jaeger/model/adjuster"
 	uiconv "github.com/jaegertracing/jaeger/model/converter/json"
 	ui "github.com/jaegertracing/jaeger/model/json"
 	"github.com/jaegertracing/jaeger/pkg/multierror"
-	"github.com/jaegertracing/jaeger/storage/dependencystore"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
 
@@ -48,10 +47,6 @@ const (
 	defaultAPIPrefix                  = "api"
 )
 
-var (
-	errNoArchiveSpanStorage = errors.New("archive span storage was not configured")
-)
-
 // HTTPHandler handles http requests
 type HTTPHandler interface {
 	RegisterRoutes(router *mux.Router)
@@ -78,23 +73,18 @@ func NewRouter() *mux.Router {
 
 // APIHandler implements the query service public API by registering routes at httpPrefix
 type APIHandler struct {
-	spanReader        spanstore.Reader
-	archiveSpanReader spanstore.Reader
-	archiveSpanWriter spanstore.Writer
-	dependencyReader  dependencystore.Reader
-	adjuster          adjuster.Adjuster
-	logger            *zap.Logger
-	queryParser       queryParser
-	basePath          string
-	apiPrefix         string
-	tracer            opentracing.Tracer
+	queryService *querysvc.QueryService
+	queryParser  queryParser
+	basePath     string
+	apiPrefix    string
+	logger       *zap.Logger
+	tracer       opentracing.Tracer
 }
 
 // NewAPIHandler returns an APIHandler
-func NewAPIHandler(spanReader spanstore.Reader, dependencyReader dependencystore.Reader, options ...HandlerOption) *APIHandler {
+func NewAPIHandler(queryService *querysvc.QueryService, options ...HandlerOption) *APIHandler {
 	aH := &APIHandler{
-		spanReader:       spanReader,
-		dependencyReader: dependencyReader,
+		queryService: queryService,
 		queryParser: queryParser{
 			traceQueryLookbackDuration: defaultTraceQueryLookbackDuration,
 			timeNow:                    time.Now,
@@ -107,9 +97,6 @@ func NewAPIHandler(spanReader spanstore.Reader, dependencyReader dependencystore
 	if aH.apiPrefix == "" {
 		aH.apiPrefix = defaultAPIPrefix
 	}
-	if aH.adjuster == nil {
-		aH.adjuster = adjuster.Sequence(StandardAdjusters...)
-	}
 	if aH.logger == nil {
 		aH.logger = zap.NewNop()
 	}
@@ -154,7 +141,7 @@ func (aH *APIHandler) route(route string, args ...interface{}) string {
 }
 
 func (aH *APIHandler) getServices(w http.ResponseWriter, r *http.Request) {
-	services, err := aH.spanReader.GetServices(r.Context())
+	services, err := aH.queryService.GetServices(r.Context())
 	if aH.handleError(w, err, http.StatusInternalServerError) {
 		return
 	}
@@ -169,7 +156,7 @@ func (aH *APIHandler) getOperationsLegacy(w http.ResponseWriter, r *http.Request
 	vars := mux.Vars(r)
 	// given how getOperationsLegacy is bound to URL route, serviceParam cannot be empty
 	service, _ := url.QueryUnescape(vars[serviceParam])
-	operations, err := aH.spanReader.GetOperations(r.Context(), service)
+	operations, err := aH.queryService.GetOperations(r.Context(), service)
 	if aH.handleError(w, err, http.StatusInternalServerError) {
 		return
 	}
@@ -187,7 +174,7 @@ func (aH *APIHandler) getOperations(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 	}
-	operations, err := aH.spanReader.GetOperations(r.Context(), service)
+	operations, err := aH.queryService.GetOperations(r.Context(), service)
 	if aH.handleError(w, err, http.StatusInternalServerError) {
 		return
 	}
@@ -212,7 +199,7 @@ func (aH *APIHandler) search(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 	} else {
-		tracesFromStorage, err = aH.spanReader.FindTraces(r.Context(), &tQuery.TraceQueryParameters)
+		tracesFromStorage, err = aH.queryService.FindTraces(r.Context(), &tQuery.TraceQueryParameters)
 		if aH.handleError(w, err, http.StatusInternalServerError) {
 			return
 		}
@@ -238,7 +225,7 @@ func (aH *APIHandler) tracesByIDs(ctx context.Context, traceIDs []model.TraceID)
 	var errors []structuredError
 	retMe := make([]*model.Trace, 0, len(traceIDs))
 	for _, traceID := range traceIDs {
-		if trace, err := trace(ctx, traceID, aH.spanReader, aH.archiveSpanReader); err != nil {
+		if trace, err := aH.queryService.GetTrace(ctx, traceID); err != nil {
 			if err != spanstore.ErrTraceNotFound {
 				return nil, nil, err
 			}
@@ -272,7 +259,7 @@ func (aH *APIHandler) dependencies(w http.ResponseWriter, r *http.Request) {
 	}
 	endTs := time.Unix(0, 0).Add(time.Duration(endTsMillis) * time.Millisecond)
 
-	dependencies, err := aH.dependencyReader.GetDependencies(endTs, lookback)
+	dependencies, err := aH.queryService.GetDependencies(endTs, lookback)
 	if aH.handleError(w, err, http.StatusInternalServerError) {
 		return
 	}
@@ -288,7 +275,7 @@ func (aH *APIHandler) convertModelToUI(trace *model.Trace, adjust bool) (*ui.Tra
 	var errors []error
 	if adjust {
 		var err error
-		trace, err = aH.adjuster.Adjust(trace)
+		trace, err = aH.queryService.Adjust(trace)
 		if err != nil {
 			errors = append(errors, err)
 		}
@@ -352,106 +339,62 @@ func (aH *APIHandler) parseTraceID(w http.ResponseWriter, r *http.Request) (mode
 }
 
 // getTrace implements the REST API /traces/{trace-id}
-func (aH *APIHandler) getTrace(w http.ResponseWriter, r *http.Request) {
-	aH.getTraceFromReaders(w, r, aH.spanReader, aH.archiveSpanReader)
-}
-
-// getTraceFromReader parses trace ID from the path, loads the trace from specified Reader,
+// It parses trace ID from the path, fetches the trace from QueryService,
 // formats it in the UI JSON format, and responds to the client.
-func (aH *APIHandler) getTraceFromReaders(
-	w http.ResponseWriter,
-	r *http.Request,
-	reader spanstore.Reader,
-	backupReader spanstore.Reader,
-) {
-	aH.withTraceFromReader(w, r, reader, backupReader, func(trace *model.Trace) {
-		var uiErrors []structuredError
-		uiTrace, uiErr := aH.convertModelToUI(trace, shouldAdjust(r))
-		if uiErr != nil {
-			uiErrors = append(uiErrors, *uiErr)
-		}
-
-		structuredRes := structuredResponse{
-			Data: []*ui.Trace{
-				uiTrace,
-			},
-			Errors: uiErrors,
-		}
-		aH.writeJSON(w, r, &structuredRes)
-	})
-}
-
-func shouldAdjust(r *http.Request) bool {
-	raw := r.FormValue("raw")
-	isRaw, _ := strconv.ParseBool(raw)
-	return !isRaw
-}
-
-// withTraceFromReader tries to load a trace from Reader and if successful
-// execute process() function passing it that trace.
-func (aH *APIHandler) withTraceFromReader(
-	w http.ResponseWriter,
-	r *http.Request,
-	reader spanstore.Reader,
-	backupReader spanstore.Reader,
-	process func(trace *model.Trace),
-) {
+func (aH *APIHandler) getTrace(w http.ResponseWriter, r *http.Request) {
 	traceID, ok := aH.parseTraceID(w, r)
 	if !ok {
 		return
 	}
-	trace, err := trace(r.Context(), traceID, reader, backupReader)
+	trace, err := aH.queryService.GetTrace(r.Context(), traceID)
 	if err == spanstore.ErrTraceNotFound {
 		aH.handleError(w, err, http.StatusNotFound)
 		return
 	}
 	if aH.handleError(w, err, http.StatusInternalServerError) {
 		return
 	}
-	process(trace)
-}
 
-func trace(
-	ctx context.Context,
-	traceID model.TraceID,
-	reader spanstore.Reader,
-	backupReader spanstore.Reader,
-) (*model.Trace, error) {
-	trace, err := reader.GetTrace(ctx, traceID)
-	if err == spanstore.ErrTraceNotFound {
-		if backupReader == nil {
-			return nil, err
-		}
-		trace, err = backupReader.GetTrace(ctx, traceID)
+	var uiErrors []structuredError
+	uiTrace, uiErr := aH.convertModelToUI(trace, shouldAdjust(r))
+	if uiErr != nil {
+		uiErrors = append(uiErrors, *uiErr)
 	}
-	return trace, err
+
+	structuredRes := structuredResponse{
+		Data: []*ui.Trace{
+			uiTrace,
+		},
+		Errors: uiErrors,
+	}
+	aH.writeJSON(w, r, &structuredRes)
+}
+
+func shouldAdjust(r *http.Request) bool {
+	raw := r.FormValue("raw")
+	isRaw, _ := strconv.ParseBool(raw)
+	return !isRaw
 }
 
 // archiveTrace implements the REST API POST:/archive/{trace-id}.
-// It reads the trace from the main Reader and saves it to archive Writer.
+// It passes the traceID to queryService.ArchiveTrace for writing.
 func (aH *APIHandler) archiveTrace(w http.ResponseWriter, r *http.Request) {
-	if aH.archiveSpanWriter == nil {
-		aH.handleError(w, errNoArchiveSpanStorage, http.StatusInternalServerError)
+	traceID, ok := aH.parseTraceID(w, r)
+	if !ok {
 		return
 	}
-	aH.withTraceFromReader(w, r, aH.spanReader, nil, func(trace *model.Trace) {
-		var writeErrors []error
-		for _, span := range trace.Spans {
-			err := aH.archiveSpanWriter.WriteSpan(span)
-			if err != nil {
-				writeErrors = append(writeErrors, err)
-			}
-		}
-		err := multierror.Wrap(writeErrors)
-		if aH.handleError(w, err, http.StatusInternalServerError) {
-			return
-		}
-		structuredRes := structuredResponse{
-			Data:   []string{}, // doens't matter, just want an empty array
-			Errors: []structuredError{},
-		}
-		aH.writeJSON(w, r, &structuredRes)
-	})
+
+	// QueryService.ArchiveTrace can now archive this traceID.
+	err := aH.queryService.ArchiveTrace(r.Context(), traceID)
+	if aH.handleError(w, err, http.StatusInternalServerError) {
+		return
+	}
+
+	structuredRes := structuredResponse{
+		Data:   []string{}, // doens't matter, just want an empty array
+		Errors: []structuredError{},
+	}
+	aH.writeJSON(w, r, &structuredRes)
 }
 
 func (aH *APIHandler) handleError(w http.ResponseWriter, err error, statusCode int) bool {