feat: Add bufferring to provisioner job logs

This should improve overall build performance, and especially under load. It removes the old `id` column on the `provisioner_job_logs` table and replaces it with an auto-incrementing big integer to preserve order. Funny enough, we never had to care about order before because inserts would at minimum be 1ms different. Now they aren't, so the order needs to be preserved.
coder · kylecarbs · Nov 7, 2022 · Nov 6, 2022 · Nov 6, 2022 · Nov 6, 2022
commit d7d4a339de98c1131f6547023fe77e9a3088a910
diff --git a/coderd/database/databasefake/databasefake.go b/coderd/database/databasefake/databasefake.go
@@ -2212,10 +2212,15 @@ func (q *fakeQuerier) InsertProvisionerJobLogs(_ context.Context, arg database.I
 	defer q.mutex.Unlock()
 
 	logs := make([]database.ProvisionerJobLog, 0)
+	id := int64(1)
+	if len(q.provisionerJobLogs) > 0 {
+		id = q.provisionerJobLogs[len(q.provisionerJobLogs)-1].ID
+	}
 	for index, output := range arg.Output {
+		id++
 		logs = append(logs, database.ProvisionerJobLog{
+			ID:        id,
 			JobID:     arg.JobID,
-			ID:        arg.ID[index],
 			CreatedAt: arg.CreatedAt[index],
 			Source:    arg.Source[index],
 			Level:     arg.Level[index],

diff --git a/coderd/database/dump.sql b/coderd/database/dump.sql
diff --git a/coderd/database/migrations/000070_provisioner_log_lines.down.sql b/coderd/database/migrations/000070_provisioner_log_lines.down.sql
diff --git a/coderd/database/migrations/000070_provisioner_log_lines.up.sql b/coderd/database/migrations/000070_provisioner_log_lines.up.sql
@@ -0,0 +1,3 @@
+ALTER TABLE provisioner_job_logs DROP COLUMN id;
+
+ALTER TABLE provisioner_job_logs ADD COLUMN id BIGSERIAL PRIMARY KEY; 
diff --git a/coderd/database/models.go b/coderd/database/models.go
diff --git a/coderd/database/queries.sql.go b/coderd/database/queries.sql.go
diff --git a/coderd/database/queries/provisionerjoblogs.sql b/coderd/database/queries/provisionerjoblogs.sql
@@ -8,15 +8,12 @@ WHERE
 	AND (
 		created_at >= @created_after
 		OR created_at <= @created_before
-	)
-ORDER BY
-	created_at DESC;
+	) ORDER BY id;
 
 -- name: InsertProvisionerJobLogs :many
 INSERT INTO
 	provisioner_job_logs
 SELECT
-	unnest(@id :: uuid [ ]) AS id,
 	@job_id :: uuid AS job_id,
 	unnest(@created_at :: timestamptz [ ]) AS created_at,
 	unnest(@source :: log_source [ ]) AS source,

diff --git a/coderd/provisionerdaemons.go b/coderd/provisionerdaemons.go
@@ -368,7 +368,6 @@ func (server *provisionerdServer) UpdateJob(ctx context.Context, request *proto.
 			if err != nil {
 				return nil, xerrors.Errorf("convert log source: %w", err)
 			}
-			insertParams.ID = append(insertParams.ID, uuid.New())
 			insertParams.CreatedAt = append(insertParams.CreatedAt, time.UnixMilli(log.CreatedAt))
 			insertParams.Level = append(insertParams.Level, logLevel)
 			insertParams.Stage = append(insertParams.Stage, log.Stage)
@@ -379,17 +378,13 @@ func (server *provisionerdServer) UpdateJob(ctx context.Context, request *proto.
 				slog.F("stage", log.Stage),
 				slog.F("output", log.Output))
 		}
-		logs, err := server.Database.InsertProvisionerJobLogs(context.Background(), insertParams)
+		_, err := server.Database.InsertProvisionerJobLogs(context.Background(), insertParams)
 		if err != nil {
 			server.Logger.Error(ctx, "failed to insert job logs", slog.F("job_id", parsedID), slog.Error(err))
 			return nil, xerrors.Errorf("insert job logs: %w", err)
 		}
 		server.Logger.Debug(ctx, "inserted job logs", slog.F("job_id", parsedID))
-		data, err := json.Marshal(provisionerJobLogsMessage{Logs: logs})
-		if err != nil {
-			return nil, xerrors.Errorf("marshal job log: %w", err)
-		}
-		err = server.Pubsub.Publish(provisionerJobLogsChannel(parsedID), data)
+		err = server.Pubsub.Publish(provisionerJobLogsChannel(parsedID), []byte("{}"))
 		if err != nil {
 			server.Logger.Error(ctx, "failed to publish job logs", slog.F("job_id", parsedID), slog.Error(err))
 			return nil, xerrors.Errorf("publish job log: %w", err)

diff --git a/coderd/provisionerjobs.go b/coderd/provisionerjobs.go
@@ -156,7 +156,7 @@ func (api *API) provisionerJobLogs(rw http.ResponseWriter, r *http.Request, job
 	ctx, wsNetConn := websocketNetConn(ctx, conn, websocket.MessageText)
 	defer wsNetConn.Close() // Also closes conn.
 
-	logIdsDone := make(map[uuid.UUID]bool)
+	logIdsDone := make(map[int64]bool)
 
 	// The Go stdlib JSON encoder appends a newline character after message write.
 	encoder := json.NewEncoder(wsNetConn)
@@ -370,8 +370,7 @@ func provisionerJobLogsChannel(jobID uuid.UUID) string {
 
 // provisionerJobLogsMessage is the message type published on the provisionerJobLogsChannel() channel
 type provisionerJobLogsMessage struct {
-	EndOfLogs bool                         `json:"end_of_logs,omitempty"`
-	Logs      []database.ProvisionerJobLog `json:"logs,omitempty"`
+	EndOfLogs bool `json:"end_of_logs,omitempty"`
 }
 
 func (api *API) followLogs(jobID uuid.UUID) (<-chan database.ProvisionerJobLog, func(), error) {
@@ -380,6 +379,7 @@ func (api *API) followLogs(jobID uuid.UUID) (<-chan database.ProvisionerJobLog,
 	var (
 		closed       = make(chan struct{})
 		bufferedLogs = make(chan database.ProvisionerJobLog, 128)
+		since        = database.Now()
 	)
 	closeSubscribe, err := api.Pubsub.Subscribe(
 		provisionerJobLogsChannel(jobID),
@@ -389,15 +389,23 @@ func (api *API) followLogs(jobID uuid.UUID) (<-chan database.ProvisionerJobLog,
 				return
 			default:
 			}
-
 			jlMsg := provisionerJobLogsMessage{}
 			err := json.Unmarshal(message, &jlMsg)
 			if err != nil {
 				logger.Warn(ctx, "invalid provisioner job log on channel", slog.Error(err))
 				return
 			}
+			logs, err := api.Database.GetProvisionerLogsByIDBetween(ctx, database.GetProvisionerLogsByIDBetweenParams{
+				JobID:        jobID,
+				CreatedAfter: since,
+			})
+			if err != nil {
+				logger.Warn(ctx, "get provisioner logs", slog.Error(err))
+				return
+			}
+			since = database.Now()
 
-			for _, log := range jlMsg.Logs {
+			for _, log := range logs {
 				select {
 				case bufferedLogs <- log:
 					logger.Debug(ctx, "subscribe buffered log", slog.F("stage", log.Stage))
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,3 @@
		ALTER TABLE provisioner_job_logs DROP COLUMN id;

		ALTER TABLE provisioner_job_logs ADD COLUMN id BIGSERIAL PRIMARY KEY;