Fix: test time (#586)

* set completedAt at the end of the test execution

* fix poller and set zero value as initial value of run

* add other important timestamps

* map new time fields to openapi spec

* fill completedAt if run fails

* fix tests

* remove StartedAt and use ServiceTriggeredAt in trace poller

* add ServiceTriggerCompletedAt field and fix runner tests

* add retry delay as config in runner to prevent flaky tests

* remove zero values from time

* generate cli http client to apply openapi changes

api/tests.yaml

@@ -87,6 +87,15 @@ components:
         createdAt:
           type: string
           format: date-time
+        serviceTriggeredAt:
+          type: string
+          format: date-time
+        serviceTriggerCompletedAt:
+          type: string
+          format: date-time
+        obtainedTraceAt:
+          type: string
+          format: date-time
         completedAt:
           type: string
           format: date-time

server/app/app.go

@@ -74,7 +74,7 @@ func (a *App) Start() error {
 	assertionRunner.Start(5)
 	defer assertionRunner.Stop()
 
-	tracePoller := executor.NewTracePoller(a.traceDB, a.db, a.config.MaxWaitTimeForTraceDuration(), subscriptionManager, assertionRunner)
+	tracePoller := executor.NewTracePoller(a.traceDB, a.db, a.config.PoolingRetryDelay(), a.config.MaxWaitTimeForTraceDuration(), subscriptionManager, assertionRunner)
 	tracePoller.Start(5) // worker count. should be configurable
 	defer tracePoller.Stop()
 

server/config/config.go

@@ -12,22 +12,37 @@ import (
 
 type (
 	Config struct {
-		PostgresConnString     string                         `mapstructure:"postgresConnString"`
-		JaegerConnectionConfig *configgrpc.GRPCClientSettings `mapstructure:"jaegerConnectionConfig"`
-		TempoConnectionConfig  *configgrpc.GRPCClientSettings `mapstructure:"tempoConnectionConfig"`
-		MaxWaitTimeForTrace    string                         `mapstructure:"maxWaitTimeForTrace"`
-		GA                     GoogleAnalytics                `mapstructure:"googleAnalytics"`
+		PostgresConnString      string                         `mapstructure:"postgresConnString"`
+		JaegerConnectionConfig  *configgrpc.GRPCClientSettings `mapstructure:"jaegerConnectionConfig"`
+		TempoConnectionConfig   *configgrpc.GRPCClientSettings `mapstructure:"tempoConnectionConfig"`
+		PoolingConfig           PoolingConfig                  `mapstructure:"poolingConfig"`
+		GA                      GoogleAnalytics                `mapstructure:"googleAnalytics"`
+		PoolingRetryDelayString string                         `mapśtructure:"poolingRetryDelay"`
 	}
 
 	GoogleAnalytics struct {
 		MeasurementID string `mapstructure:"measurementId"`
 		SecretKey     string `mapstructure:"secretKey"`
 		Enabled       bool   `mapstructure:"enabled"`
 	}
+
+	PoolingConfig struct {
+		MaxWaitTimeForTrace string `mapstructure:"maxWaitTimeForTrace"`
+		RetryDelay          string `mapstructure:"retryDelay"`
+	}
 )
 
+func (c Config) PoolingRetryDelay() time.Duration {
+	delay, err := time.ParseDuration(c.PoolingConfig.RetryDelay)
+	if err != nil {
+		return 1 * time.Second
+	}
+
+	return delay
+}
+
 func (c Config) MaxWaitTimeForTraceDuration() time.Duration {
-	maxWaitTimeForTrace, err := time.ParseDuration(c.MaxWaitTimeForTrace)
+	maxWaitTimeForTrace, err := time.ParseDuration(c.PoolingConfig.MaxWaitTimeForTrace)
 	if err != nil {
 		// use a default value
 		maxWaitTimeForTrace = 30 * time.Second

server/config/config_test.go

@@ -21,8 +21,11 @@ func TestFromFileSuccess(t *testing.T) {
 			name: "Jaeger",
 			file: "./testdata/jaeger.yaml",
 			expected: config.Config{
-				MaxWaitTimeForTrace: "1m",
-				PostgresConnString:  "host=postgres user=postgres password=postgres port=5432 sslmode=disable",
+				PoolingConfig: config.PoolingConfig{
+					MaxWaitTimeForTrace: "1m",
+					RetryDelay:          "3s",
+				},
+				PostgresConnString: "host=postgres user=postgres password=postgres port=5432 sslmode=disable",
 				JaegerConnectionConfig: &configgrpc.GRPCClientSettings{
 					Endpoint:   "jaeger-query:16685",
 					TLSSetting: configtls.TLSClientSetting{Insecure: true},
@@ -33,8 +36,11 @@ func TestFromFileSuccess(t *testing.T) {
 			name: "Tempo",
 			file: "./testdata/tempo.yaml",
 			expected: config.Config{
-				MaxWaitTimeForTrace: "1m",
-				PostgresConnString:  "host=postgres user=postgres password=postgres port=5432 sslmode=disable",
+				PoolingConfig: config.PoolingConfig{
+					MaxWaitTimeForTrace: "1m",
+					RetryDelay:          "1s",
+				},
+				PostgresConnString: "host=postgres user=postgres password=postgres port=5432 sslmode=disable",
 				TempoConnectionConfig: &configgrpc.GRPCClientSettings{
 					Endpoint:   "tempo:9095",
 					TLSSetting: configtls.TLSClientSetting{Insecure: true},

server/config/testdata/jaeger.yaml

@@ -1,4 +1,6 @@
-maxWaitTimeForTrace: 1m
+poolingConfig:
+  maxWaitTimeForTrace: 1m
+  retryDelay: 3s
 postgresConnString: "host=postgres user=postgres password=postgres port=5432 sslmode=disable"
 jaegerConnectionConfig:
   endpoint: jaeger-query:16685

server/config/testdata/tempo.yaml

@@ -1,4 +1,6 @@
-maxWaitTimeForTrace: 1m
+poolingConfig:
+  maxWaitTimeForTrace: 1m
+  retryDelay: 1s
 postgresConnString: "host=postgres user=postgres password=postgres port=5432 sslmode=disable"
 tempoConnectionConfig:
   endpoint: tempo:9095

server/executor/assertion_runner.go

@@ -75,6 +75,7 @@ func (e *defaultAssertionRunner) runAssertionsAndUpdateResult(ctx context.Contex
 	if err != nil {
 		run.State = model.RunStateFailed
 		run.LastError = err
+		run.CompletedAt = time.Now()
 		return e.db.UpdateRun(ctx, run)
 	}
 

server/executor/runner.go

@@ -108,6 +108,7 @@ func (r persistentRunner) Run(test model.Test) model.Run {
 func (r persistentRunner) processExecQueue(job execReq) {
 	run := job.run
 	run.State = model.RunStateExecuting
+	run.ServiceTriggeredAt = time.Now()
 	r.handleDBError(r.tests.UpdateRun(job.ctx, run))
 
 	response, err := r.executor.Execute(job.test, job.run.TraceID, job.run.SpanID)
@@ -118,6 +119,8 @@ func (r persistentRunner) processExecQueue(job execReq) {
 		r.handleDBError(r.tests.UpdateTestVersion(job.ctx, job.test))
 	}
 
+	run.ServiceTriggerCompletedAt = time.Now()
+
 	r.handleDBError(r.tests.UpdateRun(job.ctx, run))
 	if run.State == model.RunStateAwaitingTrace {
 		// start a new context
@@ -126,11 +129,11 @@ func (r persistentRunner) processExecQueue(job execReq) {
 }
 
 func (r persistentRunner) handleExecutionResult(run model.Run, resp model.HTTPResponse, err error) model.Run {
-	run.CompletedAt = time.Now()
 	run.Response = resp
 	if err != nil {
 		run.State = model.RunStateFailed
 		run.LastError = err
+		run.CompletedAt = time.Now()
 	} else {
 		run.State = model.RunStateAwaitingTrace
 	}
@@ -142,7 +145,7 @@ func (r persistentRunner) newTestRun() model.Run {
 		ID:        r.idGen.UUID(),
 		TraceID:   r.idGen.TraceID(),
 		SpanID:    r.idGen.SpanID(),
-		CreatedAt: time.Now(),
 		State:     model.RunStateCreated,
+		CreatedAt: time.Now(),
 	}
 }

server/executor/runner_test.go

@@ -30,7 +30,7 @@ func TestPersistentRunner(t *testing.T) {
 
 		result := f.mockDB.runs[test.ID.String()]
 		require.NotNil(t, result)
-		assert.Greater(t, result.CompletedAt.UnixNano(), result.CreatedAt.UnixNano())
+		assert.Greater(t, result.ServiceTriggerCompletedAt.UnixNano(), result.CreatedAt.UnixNano())
 
 		f.assert(t)
 	})
@@ -54,7 +54,7 @@ func TestPersistentRunner(t *testing.T) {
 		run2 := f.mockDB.runs[test2.ID.String()]
 		require.NotNil(t, run2)
 
-		assert.True(t, run1.CompletedAt.UnixNano() > run2.CompletedAt.UnixNano(), "test1 did not complete after test2")
+		assert.Greater(t, run1.ServiceTriggerCompletedAt.UnixNano(), run2.ServiceTriggerCompletedAt.UnixNano(), "test1 did not complete after test2")
 	})
 
 }

server/executor/trace_poller.go

@@ -30,11 +30,11 @@ type TraceFetcher interface {
 func NewTracePoller(
 	tf TraceFetcher,
 	tests model.Repository,
+	retryDelay time.Duration,
 	maxWaitTimeForTrace time.Duration,
 	subscriptionManager *subscription.Manager,
 	assertionRunner AssertionRunner,
 ) PersistentTracePoller {
-	retryDelay := 1 * time.Second
 	maxTracePollRetry := int(math.Ceil(float64(maxWaitTimeForTrace) / float64(retryDelay)))
 	return tracePoller{
 		tests:               tests,
@@ -140,6 +140,7 @@ func (tp tracePoller) processJob(job tracePollReq) {
 
 	run.Trace = augmentData(&trace, run.Response)
 	run.State = model.RunStateAwaitingTestResults
+	run.ObtainedTraceAt = time.Now()
 
 	fmt.Printf("completed polling result %s after %d times, number of spans: %d \n", job.run.ID, job.count, len(run.Trace.Flat))
 
@@ -201,7 +202,7 @@ func (tp tracePoller) donePollingTraces(job tracePollReq, trace traces.Trace) bo
 func (tp tracePoller) handleTraceDBError(job tracePollReq, err error) {
 	run := job.run
 	if errors.Is(err, tracedb.ErrTraceNotFound) {
-		if time.Since(run.CompletedAt) < tp.maxWaitTimeForTrace {
+		if time.Since(run.ServiceTriggeredAt) < tp.maxWaitTimeForTrace {
 			tp.requeue(job)
 			return
 		}
@@ -214,6 +215,7 @@ func (tp tracePoller) handleTraceDBError(job tracePollReq, err error) {
 
 	run.State = model.RunStateFailed
 	run.LastError = err
+	run.CompletedAt = time.Now()
 
 	tp.handleDBError(tp.tests.UpdateRun(job.ctx, run))