Merge #88395 #88504 #88509

88395: changefeedccl: Do not block on file size based flushes r=miretskiy a=miretskiy

Prior to this change, cloud storage sink trigger
file sized based flush whenever new row would
would push the file size beyond configured threshold.

This had the effect of singificantly reducing the throughput whenever such event occured -- no additional events could be added to cloud storage sink, while the previus flush was active.

This is not necessary.  Cloud storage sink can trigger file based flushes asynchronously.  The only requirement is that if a real, non file based, flush arrives, or if we need to emit resolved timestamps, then we must wait for all of the active flush requests to complete.

In addition, because every event added to cloud sink has associate allocation, which is released when file is written out, performing flushes asynchronously is safe with respect to memory usage and accounting.

Release note (enterprise change): Changefeeds, using cloud storage sink, now have better throughput.
Release justification: performance fix

88504: kvserver: add storage time-series metrics for level size and score r=sumeerbhola,jbowens a=nicktrav

Currently, the only way to infer the compaction score and heuristics is to use the LSM printout from the logs (emitted once every ten minutes), or to call the `/debug/lsm` endpoint manually, and track values over time. This makes it difficult to debug issues retroactively.

Add two new sets of per-LSM-level time-series metrics for level size and level score. These new metrics have names of the form `storage.$LEVEL-level-{size,score}`.

Closes #88415.

Release note (ops change): Adds two new sets of per-LSM-level time-series metrics, one for level size and another for level score. These metrics are of the form `storage.$LEVEL-level-{size,score}`.

88509: ui: insights overview rename 'execution id' to 'latest execution id' r=j82w a=j82w

The insights only shows the latest execution id per a fingerprint. Renaming the column will avoid confusion where user might expect multiple execution ids for the same fingerprint. 

After showing the new column name.
<img width="1131" alt="Screen Shot 2022-09-22 at 4 15 12 PM" src="https://user-images.githubusercontent.com/8868107/191842420-92877707-ed8d-4eee-af00-052fd1544516.png">
<img width="1457" alt="Screen Shot 2022-09-22 at 4 14 22 PM" src="https://user-images.githubusercontent.com/8868107/191842732-8c69099d-f18a-434f-b6c5-4ce231f364cc.png">

closes #88456

Release justification: Category 2: Bug fixes and
low-risk updates to new functionality

Release note: (ui change): Rename insights overview table column 'execution id' to 'latest execution id'. This will help avoid confusion since the ui only shows the latest id per fingerprint.

Co-authored-by: Yevgeniy Miretskiy <yevgeniy@cockroachlabs.com>
Co-authored-by: Nick Travers <travers@cockroachlabs.com>
Co-authored-by: j82w <jwilley@cockroachlabs.com>

pkg/ccl/changefeedccl/sink_cloudstorage.go

@@ -19,6 +19,7 @@ import (
 	"net/url"
 	"path/filepath"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -27,6 +28,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/kvevent"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
@@ -304,6 +306,11 @@ type cloudStorageSink struct {
 	dataFilePartition string
 	prevFilename      string
 	metrics           metricsRecorder
+
+	asyncFlushActive bool
+	flushCtx         context.Context
+	flushGroup       sync.WaitGroup
+	flushErr         atomic.Value
 }
 
 const sinkCompressionGzip = "gzip"
@@ -364,8 +371,11 @@ func makeCloudStorageSink(
 		partitionFormat:   defaultPartitionFormat,
 		timestampOracle:   timestampOracle,
 		// TODO(dan,ajwerner): Use the jobs framework's session ID once that's available.
-		jobSessionID: sessID,
-		topicNamer:   tn,
+		jobSessionID:     sessID,
+		topicNamer:       tn,
+		asyncFlushActive: enableAsyncFlush.Get(&settings.SV),
+		// TODO (yevgeniy): Consider adding ctx to Dial method instead.
+		flushCtx: ctx,
 	}
 
 	if partitionFormat := u.consumeParam(changefeedbase.SinkParamPartitionFormat); partitionFormat != "" {
@@ -503,6 +513,13 @@ func (s *cloudStorageSink) EmitResolvedTimestamp(
 	if err != nil {
 		return err
 	}
+
+	// Wait for previously issued async flush requests to complete
+	// before we write  resolved time stamp file.
+	if err := s.waitAsyncFlush(); err != nil {
+		return errors.Wrapf(err, "while emitting resolved timestamp")
+	}
+
 	// Don't need to copy payload because we never buffer it anywhere.
 
 	part := resolved.GoTime().Format(s.partitionFormat)
@@ -572,29 +589,44 @@ func (s *cloudStorageSink) Flush(ctx context.Context) error {
 	// for an overview of the naming convention and proof of correctness.
 	s.dataFileTs = cloudStorageFormatTime(s.timestampOracle.inclusiveLowerBoundTS())
 	s.dataFilePartition = s.timestampOracle.inclusiveLowerBoundTS().GoTime().Format(s.partitionFormat)
-	return nil
+	return s.waitAsyncFlush()
 }
 
-// file should not be used after flushing.
-func (s *cloudStorageSink) flushFile(ctx context.Context, file *cloudStorageSinkFile) error {
-	defer file.alloc.Release(ctx)
+// enableAsyncFlush controls async flushing behavior for this sink.
+var enableAsyncFlush = settings.RegisterBoolSetting(
+	settings.TenantWritable,
+	"changefeed.cloudstorage.async_flush.enabled",
+	"enable async flushing",
+	true,
+)
 
-	if file.rawSize == 0 {
-		// This method shouldn't be called with an empty file, but be defensive
-		// about not writing empty files anyway.
-		return nil
+// waitAsyncFlush waits until all async flushes complete.
+func (s *cloudStorageSink) waitAsyncFlush() error {
+	s.flushGroup.Wait()
+	if v := s.flushErr.Load(); v != nil {
+		return v.(error)
 	}
+	return nil
+}
 
-	if file.codec != nil {
-		if err := file.codec.Close(); err != nil {
+// flushFile flushes file to the cloud storage.
+// file should not be used after flushing.
+func (s *cloudStorageSink) flushFile(ctx context.Context, file *cloudStorageSinkFile) error {
+	asyncFlushEnabled := enableAsyncFlush.Get(&s.settings.SV)
+	if s.asyncFlushActive && !asyncFlushEnabled {
+		// Async flush behavior was turned off --  drain any active flush requests
+		// before flushing this file.
+		if err := s.waitAsyncFlush(); err != nil {
 			return err
 		}
 	}
+	s.asyncFlushActive = asyncFlushEnabled
 
 	// We use this monotonically increasing fileID to ensure correct ordering
 	// among files emitted at the same timestamp during the same job session.
 	fileID := s.fileID
 	s.fileID++
+
 	// Pad file ID to maintain lexical ordering among files from the same sink.
 	// Note that we use `-` here to delimit the filename because we want
 	// `%d.RESOLVED` files to lexicographically succeed data files that have the
@@ -606,19 +638,60 @@ func (s *cloudStorageSink) flushFile(ctx context.Context, file *cloudStorageSink
 			"precedes a file emitted before: %s", filename, s.prevFilename)
 	}
 	s.prevFilename = filename
-	compressedBytes := file.buf.Len()
-	if err := cloud.WriteFile(ctx, s.es, filepath.Join(s.dataFilePartition, filename), bytes.NewReader(file.buf.Bytes())); err != nil {
+	dest := filepath.Join(s.dataFilePartition, filename)
+
+	if !asyncFlushEnabled {
+		return file.flushToStorage(ctx, s.es, dest, s.metrics)
+	}
+
+	s.flushGroup.Add(1)
+	go func() {
+		defer s.flushGroup.Done()
+		// NB: must use s.flushCtx; ctx may be short lived (i.e. cancelled).
+		if err := file.flushToStorage(s.flushCtx, s.es, dest, s.metrics); err != nil {
+			log.Errorf(ctx, "error flushing file to storage: %s", err)
+			// We must use the same type for error we store in flushErr.
+			s.flushErr.CompareAndSwap(nil, &flushError{error: err})
+		}
+	}()
+	return nil
+}
+
+type flushError struct {
+	error
+}
+
+// flushToStorage writes out file into external storage into 'dest'.
+func (f *cloudStorageSinkFile) flushToStorage(
+	ctx context.Context, es cloud.ExternalStorage, dest string, m metricsRecorder,
+) error {
+	defer f.alloc.Release(ctx)
+
+	if f.rawSize == 0 {
+		// This method shouldn't be called with an empty file, but be defensive
+		// about not writing empty files anyway.
+		return nil
+	}
+
+	if f.codec != nil {
+		if err := f.codec.Close(); err != nil {
+			return err
+		}
+	}
+
+	compressedBytes := f.buf.Len()
+	if err := cloud.WriteFile(ctx, es, dest, bytes.NewReader(f.buf.Bytes())); err != nil {
 		return err
 	}
-	s.metrics.recordEmittedBatch(file.created, file.numMessages, file.oldestMVCC, file.rawSize, compressedBytes)
+	m.recordEmittedBatch(f.created, f.numMessages, f.oldestMVCC, f.rawSize, compressedBytes)
 
 	return nil
 }
 
 // Close implements the Sink interface.
 func (s *cloudStorageSink) Close() error {
 	s.files = nil
-	return s.es.Close()
+	return errors.CombineErrors(s.waitAsyncFlush(), s.es.Close())
 }
 
 // Dial implements the Sink interface.