Smol refactor

Signed-off-by: Danny Kopping <danny@coder.com>
coder · dannykopping · Aug 21, 2024 · Aug 7, 2024 · Aug 13, 2024 · Aug 13, 2024
commit 803a9d47458f62f1b5398d80d43c9fdf968c4b6e
diff --git a/coderd/database/dbtime/dbtime.go b/coderd/database/dbtime/dbtime.go
@@ -10,5 +10,5 @@ func Now() time.Time {
 // Time returns a time compatible with Postgres. Postgres only stores dates with
 // microsecond precision.
 func Time(t time.Time) time.Time {
-	return t.Round(time.Microsecond)
-}
+	return t.UTC().Round(time.Microsecond)
+}
diff --git a/...s/000242_provisioner_job_timings.down.sql → ...s/000244_provisioner_job_timings.down.sql b/...s/000242_provisioner_job_timings.down.sql → ...s/000244_provisioner_job_timings.down.sql
diff --git a/...ons/000242_provisioner_job_timings.up.sql → ...ons/000244_provisioner_job_timings.up.sql b/...ons/000242_provisioner_job_timings.up.sql → ...ons/000244_provisioner_job_timings.up.sql
diff --git a/provisioner/terraform/executor.go b/provisioner/terraform/executor.go
@@ -21,7 +21,6 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	"golang.org/x/xerrors"
 
-	"github.com/coder/coder/v2/coderd/database"
 	"github.com/coder/coder/v2/coderd/tracing"
 	"github.com/coder/coder/v2/provisionersdk/proto"
 )
@@ -36,6 +35,8 @@ type executor struct {
 	// cachePath and workdir must not be used by multiple processes at once.
 	cachePath string
 	workdir   string
+	// used to capture execution times at various stages
+	timings *timingAggregator
 }
 
 func (e *executor) basicEnv() []string {
@@ -254,8 +255,7 @@ func (e *executor) plan(ctx, killCtx context.Context, env, vars []string, logr l
 		args = append(args, "-var", variable)
 	}
 
-	timingsAgg := newTimingsAggregator(database.ProvisionerJobTimingStagePlan)
-	outWriter, doneOut := e.provisionLogWriter(logr, timingsAgg)
+	outWriter, doneOut := e.provisionLogWriter(logr)
 	errWriter, doneErr := logWriter(logr, proto.LogLevel_ERROR)
 	defer func() {
 		_ = outWriter.Close()
@@ -277,7 +277,7 @@ func (e *executor) plan(ctx, killCtx context.Context, env, vars []string, logr l
 		Parameters:            state.Parameters,
 		Resources:             state.Resources,
 		ExternalAuthProviders: state.ExternalAuthProviders,
-		Timings:               timingsAgg.aggregate(),
+		Timings:               e.timings.aggregate(),
 	}, nil
 }
 
@@ -404,8 +404,7 @@ func (e *executor) apply(
 		getPlanFilePath(e.workdir),
 	}
 
-	timingsAgg := newTimingsAggregator(database.ProvisionerJobTimingStageApply)
-	outWriter, doneOut := e.provisionLogWriter(logr, timingsAgg)
+	outWriter, doneOut := e.provisionLogWriter(logr)
 	errWriter, doneErr := logWriter(logr, proto.LogLevel_ERROR)
 	defer func() {
 		_ = outWriter.Close()
@@ -433,7 +432,7 @@ func (e *executor) apply(
 		Resources:             state.Resources,
 		ExternalAuthProviders: state.ExternalAuthProviders,
 		State:                 stateContent,
-		Timings:               timingsAgg.aggregate(),
+		Timings:               e.timings.aggregate(),
 	}, nil
 }
 
@@ -547,15 +546,15 @@ func readAndLog(sink logSink, r io.Reader, done chan<- any, level proto.LogLevel
 // provisionLogWriter creates a WriteCloser that will log each JSON formatted terraform log.  The WriteCloser must be
 // closed by the caller to end logging, after which the returned channel will be closed to indicate that logging of the
 // written data has finished.  Failure to close the WriteCloser will leak a goroutine.
-func (e *executor) provisionLogWriter(sink logSink, timings *timingsAggregator) (io.WriteCloser, <-chan any) {
+func (e *executor) provisionLogWriter(sink logSink) (io.WriteCloser, <-chan any) {
 	r, w := io.Pipe()
 	done := make(chan any)
 
-	go e.provisionReadAndLog(sink, r, timings, done)
+	go e.provisionReadAndLog(sink, r, done)
 	return w, done
 }
 
-func (e *executor) provisionReadAndLog(sink logSink, r io.Reader, timings *timingsAggregator, done chan<- any) {
+func (e *executor) provisionReadAndLog(sink logSink, r io.Reader, done chan<- any) {
 	defer close(done)
 	scanner := bufio.NewScanner(r)
 	for scanner.Scan() {
@@ -588,13 +587,13 @@ func (e *executor) provisionReadAndLog(sink logSink, r io.Reader, timings *timin
 		logLevel := convertTerraformLogLevel(log.Level, sink)
 		sink.ProvisionLog(logLevel, log.Message)
 
-		ts, te, err := extractTimingsEntry(log)
+		ts, span, err := extractTimingSpan(log)
 		if err != nil {
 			e.logger.Debug(context.Background(), "failed to extract timings entry from log line",
 				slog.F("line", log.Message), slog.Error(err))
 		} else {
 			// Only ingest valid timings.
-			timings.ingest(ts, te)
+			e.timings.ingest(ts, span)
 		}
 
 		// If the diagnostic is provided, let's provide a bit more info!
@@ -608,13 +607,13 @@ func (e *executor) provisionReadAndLog(sink logSink, r io.Reader, timings *timin
 	}
 }
 
-func extractTimingsEntry(log terraformProvisionLog) (time.Time, *timingsEntry, error) {
+func extractTimingSpan(log terraformProvisionLog) (time.Time, *timingSpan, error) {
 	// Input is not well-formed, bail out.
 	if log.Type == "" {
 		return time.Time{}, nil, xerrors.Errorf("invalid type: %q", log.Type)
 	}
 
-	typ := logType(log.Type)
+	typ := timingKind(log.Type)
 	if !typ.Valid() {
 		return time.Time{}, nil, xerrors.Errorf("invalid type: %q", log.Type)
 	}
@@ -624,9 +623,8 @@ func extractTimingsEntry(log terraformProvisionLog) (time.Time, *timingsEntry, e
 		// TODO: log
 		ts = time.Now()
 	}
-	ts = ts.UTC()
 
-	return ts, &timingsEntry{
+	return ts, &timingSpan{
 		kind:     typ,
 		action:   log.Hook.Action,
 		provider: log.Hook.Resource.Provider,

diff --git a/provisioner/terraform/provision.go b/provisioner/terraform/provision.go
@@ -73,7 +73,7 @@ func (s *server) Plan(
 	defer cancel()
 	defer kill()
 
-	e := s.executor(sess.WorkDirectory)
+	e := s.executor(sess.WorkDirectory, database.ProvisionerJobTimingStagePlan)
 	if err := e.checkMinVersion(ctx); err != nil {
 		return provisionersdk.PlanErrorf(err.Error())
 	}
@@ -104,18 +104,20 @@ func (s *server) Plan(
 
 	s.logger.Debug(ctx, "running initialization")
 
-	timingsAgg := newTimingsAggregator(database.ProvisionerJobTimingStageInit)
-	timingsAgg.ingest(createInitTimingsEvent(initStart))
+	// The JSON output of `terraform init` doesn't include discrete fields for capturing timings of each plugin,
+	// so we capture the whole init process.
+	initTimings := newTimingAggregator(database.ProvisionerJobTimingStageInit)
+	initTimings.ingest(createInitTimingsEvent(initStart))
 
 	err = e.init(ctx, killCtx, sess)
 	if err != nil {
-		timingsAgg.ingest(createInitTimingsEvent(initErrored))
+		initTimings.ingest(createInitTimingsEvent(initErrored))
 
 		s.logger.Debug(ctx, "init failed", slog.Error(err))
 		return provisionersdk.PlanErrorf("initialize terraform: %s", err)
 	}
 
-	timingsAgg.ingest(createInitTimingsEvent(initComplete))
+	initTimings.ingest(createInitTimingsEvent(initComplete))
 
 	s.logger.Debug(ctx, "ran initialization")
 
@@ -137,12 +139,12 @@ func (s *server) Plan(
 		return provisionersdk.PlanErrorf(err.Error())
 	}
 
-	resp.Timings = append(resp.Timings, timingsAgg.aggregate()...)
+	resp.Timings = append(resp.Timings, initTimings.aggregate()...)
 	return resp
 }
 
-func createInitTimingsEvent(event logType) (time.Time, *timingsEntry) {
-	return dbtime.Now(), &timingsEntry{
+func createInitTimingsEvent(event timingKind) (time.Time, *timingSpan) {
+	return dbtime.Now(), &timingSpan{
 		kind:     event,
 		action:   "initialize terraform",
 		provider: "terraform",
@@ -159,7 +161,7 @@ func (s *server) Apply(
 	defer cancel()
 	defer kill()
 
-	e := s.executor(sess.WorkDirectory)
+	e := s.executor(sess.WorkDirectory, database.ProvisionerJobTimingStageApply)
 	if err := e.checkMinVersion(ctx); err != nil {
 		return provisionersdk.ApplyErrorf(err.Error())
 	}

diff --git a/provisioner/terraform/serve.go b/provisioner/terraform/serve.go
@@ -12,6 +12,8 @@ import (
 	"golang.org/x/xerrors"
 
 	"cdr.dev/slog"
+
+	"github.com/coder/coder/v2/coderd/database"
 	"github.com/coder/coder/v2/coderd/unhanger"
 	"github.com/coder/coder/v2/provisionersdk"
 )
@@ -138,13 +140,14 @@ func (s *server) startTrace(ctx context.Context, name string, opts ...trace.Span
 	))...)
 }
 
-func (s *server) executor(workdir string) *executor {
+func (s *server) executor(workdir string, stage database.ProvisionerJobTimingStage) *executor {
 	return &executor{
 		server:     s,
 		mut:        s.execMut,
 		binaryPath: s.binaryPath,
 		cachePath:  s.cachePath,
 		workdir:    workdir,
 		logger:     s.logger.Named("executor"),
+		timings:    newTimingAggregator(stage),
 	}
 }