[v2][query] Apply "Max Clock Skew Adjust" setting

Apply the `max_clock_skew_adjust` setting to the query service during
initialization.

The maximum clock skew adjustment setting for the query component is
properly read during initialization, but not applied in the v2 server in
the end. This causes it to always default to 1 second, regardless of the
configuration value.

Signed-off-by: Dominik Nakamura <dnaka91@gmail.com>

cmd/jaeger/config.yaml

@@ -32,6 +32,9 @@ extensions:
       traces_archive: another_store
     ui:
       config_file: ./cmd/jaeger/config-ui.json
+    # The maximum duration that is considered for clock skew adjustments.
+    # Defaults to 1 second, set to 0s to disable.
+    #max_clock_skew_adjust: 1s
 
   jaeger_storage:
     backends:

cmd/jaeger/internal/all-in-one.yaml

@@ -20,6 +20,7 @@ extensions:
   jaeger_query:
     storage:
       traces: some_storage
+    max_clock_skew_adjust: 1s
 
   jaeger_storage:
     backends:

cmd/jaeger/internal/extension/jaegerquery/server.go

@@ -95,8 +95,12 @@ func (s *server) Start(ctx context.Context, host component.Host) error {
 		return fmt.Errorf("cannot create dependencies reader: %w", err)
 	}
 
-	var opts querysvc.QueryServiceOptions
-	var v2opts v2querysvc.QueryServiceOptions
+	opts := querysvc.QueryServiceOptions{
+		MaxClockSkewAdjust: s.config.MaxClockSkewAdjust,
+	}
+	v2opts := v2querysvc.QueryServiceOptions{
+		MaxClockSkewAdjust: s.config.MaxClockSkewAdjust,
+	}
 	if err := s.addArchiveStorage(&opts, &v2opts, host); err != nil {
 		return err
 	}

cmd/query/app/flags.go

@@ -23,9 +23,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"
-	v2adjuster "github.com/jaegertracing/jaeger/cmd/query/app/querysvc/v2/adjuster"
 	v2querysvc "github.com/jaegertracing/jaeger/cmd/query/app/querysvc/v2/querysvc"
-	"github.com/jaegertracing/jaeger/model/adjuster"
 	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
 	"github.com/jaegertracing/jaeger/pkg/tenancy"
@@ -47,6 +45,10 @@ const (
 	queryEnableTracing         = "query.enable-tracing"
 )
 
+const (
+	defaultMaxClockSkewAdjust = time.Second
+)
+
 var tlsGRPCFlagsConfig = tlscfg.ServerFlagsConfig{
 	Prefix: "query.grpc",
 }
@@ -94,7 +96,7 @@ func AddFlags(flagSet *flag.FlagSet) {
 	flagSet.Bool(queryLogStaticAssetsAccess, false, "Log when static assets are accessed (for debugging)")
 	flagSet.String(queryUIConfig, "", "The path to the UI configuration file in JSON format")
 	flagSet.Bool(queryTokenPropagation, false, "Allow propagation of bearer token to be used by storage plugins")
-	flagSet.Duration(queryMaxClockSkewAdjust, 0, "The maximum delta by which span timestamps may be adjusted in the UI due to clock skew; set to 0s to disable clock skew adjustments")
+	flagSet.Duration(queryMaxClockSkewAdjust, defaultMaxClockSkewAdjust, "The maximum delta by which span timestamps may be adjusted in the UI due to clock skew; set to 0s to disable clock skew adjustments")
 	flagSet.Bool(queryEnableTracing, false, "Enables emitting jaeger-query traces")
 	tlsGRPCFlagsConfig.AddFlags(flagSet)
 	tlsHTTPFlagsConfig.AddFlags(flagSet)
@@ -140,18 +142,20 @@ func (qOpts *QueryOptions) InitFromViper(v *viper.Viper, logger *zap.Logger) (*Q
 
 // BuildQueryServiceOptions creates a QueryServiceOptions struct with appropriate adjusters and archive config
 func (qOpts *QueryOptions) BuildQueryServiceOptions(storageFactory storage.BaseFactory, logger *zap.Logger) *querysvc.QueryServiceOptions {
-	opts := &querysvc.QueryServiceOptions{}
+	opts := &querysvc.QueryServiceOptions{
+		MaxClockSkewAdjust: qOpts.MaxClockSkewAdjust,
+	}
 	if !opts.InitArchiveStorage(storageFactory, logger) {
 		logger.Info("Archive storage not initialized")
 	}
 
-	opts.Adjuster = adjuster.Sequence(querysvc.StandardAdjusters(qOpts.MaxClockSkewAdjust)...)
-
 	return opts
 }
 
 func (qOpts *QueryOptions) BuildQueryServiceOptionsV2(storageFactory storage.BaseFactory, logger *zap.Logger) *v2querysvc.QueryServiceOptions {
-	opts := &v2querysvc.QueryServiceOptions{}
+	opts := &v2querysvc.QueryServiceOptions{
+		MaxClockSkewAdjust: qOpts.MaxClockSkewAdjust,
+	}
 
 	ar, aw := v1adapter.InitializeArchiveStorage(storageFactory, logger)
 	if ar != nil && aw != nil {
@@ -161,8 +165,6 @@ func (qOpts *QueryOptions) BuildQueryServiceOptionsV2(storageFactory storage.Bas
 		logger.Info("Archive storage not initialized")
 	}
 
-	opts.Adjuster = v2adjuster.Sequence(v2adjuster.StandardAdjusters(qOpts.MaxClockSkewAdjust)...)
-
 	return opts
 }
 
@@ -197,6 +199,7 @@ func mapHTTPHeaderToOTELHeaders(h http.Header) map[string]configopaque.String {
 
 func DefaultQueryOptions() QueryOptions {
 	return QueryOptions{
+		MaxClockSkewAdjust: defaultMaxClockSkewAdjust,
 		HTTP: confighttp.ServerConfig{
 			Endpoint: ports.PortToHostPort(ports.QueryHTTP),
 		},

cmd/query/app/flags_test.go

@@ -94,9 +94,9 @@ func TestBuildQueryServiceOptions(t *testing.T) {
 
 	qSvcOpts := qOpts.BuildQueryServiceOptions(&mocks.Factory{}, zap.NewNop())
 	assert.NotNil(t, qSvcOpts)
-	assert.NotNil(t, qSvcOpts.Adjuster)
 	assert.Nil(t, qSvcOpts.ArchiveSpanReader)
 	assert.Nil(t, qSvcOpts.ArchiveSpanWriter)
+	assert.Equal(t, defaultMaxClockSkewAdjust, qSvcOpts.MaxClockSkewAdjust)
 
 	comboFactory := struct {
 		*mocks.Factory
@@ -111,9 +111,9 @@ func TestBuildQueryServiceOptions(t *testing.T) {
 
 	qSvcOpts = qOpts.BuildQueryServiceOptions(comboFactory, zap.NewNop())
 	assert.NotNil(t, qSvcOpts)
-	assert.NotNil(t, qSvcOpts.Adjuster)
 	assert.NotNil(t, qSvcOpts.ArchiveSpanReader)
 	assert.NotNil(t, qSvcOpts.ArchiveSpanWriter)
+	assert.Equal(t, defaultMaxClockSkewAdjust, qSvcOpts.MaxClockSkewAdjust)
 }
 
 func TestBuildQueryServiceOptionsV2(t *testing.T) {
@@ -124,9 +124,9 @@ func TestBuildQueryServiceOptionsV2(t *testing.T) {
 
 	qSvcOpts := qOpts.BuildQueryServiceOptionsV2(&mocks.Factory{}, zap.NewNop())
 	assert.NotNil(t, qSvcOpts)
-	assert.NotNil(t, qSvcOpts.Adjuster)
 	assert.Nil(t, qSvcOpts.ArchiveTraceReader)
 	assert.Nil(t, qSvcOpts.ArchiveTraceWriter)
+	assert.Equal(t, defaultMaxClockSkewAdjust, qSvcOpts.MaxClockSkewAdjust)
 
 	comboFactory := struct {
 		*mocks.Factory
@@ -141,9 +141,9 @@ func TestBuildQueryServiceOptionsV2(t *testing.T) {
 
 	qSvcOpts = qOpts.BuildQueryServiceOptionsV2(comboFactory, zap.NewNop())
 	assert.NotNil(t, qSvcOpts)
-	assert.NotNil(t, qSvcOpts.Adjuster)
 	assert.NotNil(t, qSvcOpts.ArchiveTraceReader)
 	assert.NotNil(t, qSvcOpts.ArchiveTraceWriter)
+	assert.Equal(t, defaultMaxClockSkewAdjust, qSvcOpts.MaxClockSkewAdjust)
 }
 
 func TestBuildQueryServiceOptionsV2_NoArchiveStorage(t *testing.T) {

cmd/query/app/querysvc/query_service.go

@@ -29,7 +29,8 @@ const (
 type QueryServiceOptions struct {
 	ArchiveSpanReader spanstore.Reader
 	ArchiveSpanWriter spanstore.Writer
-	Adjuster          adjuster.Adjuster
+	// MaxClockSkewAdjust is the maximum duration by which to adjust a span.
+	MaxClockSkewAdjust time.Duration
 }
 
 // StorageCapabilities is a feature flag for query service
@@ -44,7 +45,9 @@ type StorageCapabilities struct {
 type QueryService struct {
 	spanReader       spanstore.Reader
 	dependencyReader depstore.Reader
-	options          QueryServiceOptions
+	// Adjuster is used to adjust traces before they are returned to the client.
+	adjuster adjuster.Adjuster
+	options  QueryServiceOptions
 }
 
 // GetTraceParameters defines the parameters for querying a single trace from the query service.
@@ -70,12 +73,10 @@ func NewQueryService(traceReader tracestore.Reader, dependencyReader depstore.Re
 	qsvc := &QueryService{
 		spanReader:       spanReader,
 		dependencyReader: dependencyReader,
+		adjuster:         adjuster.Sequence(StandardAdjusters(defaultMaxClockSkewAdjust)...),
 		options:          options,
 	}
 
-	if qsvc.options.Adjuster == nil {
-		qsvc.options.Adjuster = adjuster.Sequence(StandardAdjusters(defaultMaxClockSkewAdjust)...)
-	}
 	return qsvc
 }
 
@@ -146,7 +147,7 @@ func (qs QueryService) ArchiveTrace(ctx context.Context, query spanstore.GetTrac
 
 // Adjust applies adjusters to the trace.
 func (qs QueryService) adjust(trace *model.Trace) {
-	qs.options.Adjuster.Adjust(trace)
+	qs.adjuster.Adjust(trace)
 }
 
 // GetDependencies implements dependencystore.Reader.GetDependencies

cmd/query/app/querysvc/v2/querysvc/service.go

@@ -21,19 +21,14 @@ import (
 
 var errNoArchiveSpanStorage = errors.New("archive span storage was not configured")
 
-const (
-	defaultMaxClockSkewAdjust = time.Second
-)
-
 // QueryServiceOptions holds the configuration options for the query service.
 type QueryServiceOptions struct {
 	// ArchiveTraceReader is used to read archived traces from the storage.
 	ArchiveTraceReader tracestore.Reader
 	// ArchiveTraceWriter is used to write traces to the archive storage.
 	ArchiveTraceWriter tracestore.Writer
-	// Adjuster is used to adjust traces before they are returned to the client.
-	// If not set, the default adjuster will be used.
-	Adjuster adjuster.Adjuster
+	// MaxClockSkewAdjust is the maximum duration by which to adjust a span.
+	MaxClockSkewAdjust time.Duration
 }
 
 // StorageCapabilities is a feature flag for query service
@@ -48,7 +43,9 @@ type StorageCapabilities struct {
 type QueryService struct {
 	traceReader      tracestore.Reader
 	dependencyReader depstore.Reader
-	options          QueryServiceOptions
+	// Adjuster is used to adjust traces before they are returned to the client.
+	adjuster adjuster.Adjuster
+	options  QueryServiceOptions
 }
 
 // GetTraceParams defines the parameters for retrieving traces using the GetTraces function.
@@ -76,13 +73,12 @@ func NewQueryService(
 	qsvc := &QueryService{
 		traceReader:      traceReader,
 		dependencyReader: dependencyReader,
-		options:          options,
+		adjuster: adjuster.Sequence(
+			adjuster.StandardAdjusters(options.MaxClockSkewAdjust)...,
+		),
+		options: options,
 	}
 
-	if qsvc.options.Adjuster == nil {
-		qsvc.options.Adjuster = adjuster.Sequence(
-			adjuster.StandardAdjusters(defaultMaxClockSkewAdjust)...)
-	}
 	return qsvc
 }
 
@@ -191,7 +187,7 @@ func (qs QueryService) receiveTraces(
 		}
 		for _, trace := range traces {
 			if !rawTraces {
-				qs.options.Adjuster.Adjust(trace)
+				qs.adjuster.Adjust(trace)
 			}
 			jptrace.SpanIter(trace)(func(_ jptrace.SpanIterPos, span ptrace.Span) bool {
 				foundTraceIDs[span.TraceID()] = struct{}{}

cmd/query/app/querysvc/v2/querysvc/service_test.go

@@ -60,8 +60,8 @@ func withArchiveTraceWriter() testOption {
 }
 
 func withAdjuster(adj adjuster.Adjuster) testOption {
-	return func(_ *testQueryService, options *QueryServiceOptions) {
-		options.Adjuster = adj
+	return func(tqs *testQueryService, _ *QueryServiceOptions) {
+		tqs.queryService.adjuster = adj
 	}
 }