Add a cassandra-backed storage option #246

shs96c · 2025-05-06T13:39:16Z

No description provided.

EdSchouten · 2025-05-14T11:38:19Z

pkg/blobstore/cassandra/README.md

+The tables that are required can be set up using command similar to the
+below in a `cqlsh` session:
+
+```shell


s/shell/sql/ or something?

Checking the supported languages, I think sql is probably right.

EdSchouten · 2025-05-14T11:39:34Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+	"github.com/buildbarn/bb-storage/pkg/blobstore"
+	"github.com/buildbarn/bb-storage/pkg/blobstore/buffer"
+	"github.com/buildbarn/bb-storage/pkg/blobstore/slicing"
+	"github.com/buildbarn/bb-storage/pkg/capabilities"
+	bbdigest "github.com/buildbarn/bb-storage/pkg/digest"
+	"github.com/buildbarn/bb-storage/pkg/util"
+
+	remoteexecution "github.com/bazelbuild/remote-apis/build/bazel/remote/execution/v2"
+	"github.com/gocql/gocql"
+	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"


For consistency with the rest of the code base, please group imports by second level domain. So first comes github.com, then an empty line, and then golang.org.

Is there an automated formatter for this? Or suggested IJ settings?

EdSchouten · 2025-05-14T11:40:36Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+	segmentWriteHist = prometheus.NewHistogram(prometheus.HistogramOpts{
+		Namespace: "buildbarn",
+		Subsystem: "storage",
+		Name:      "cassandra_segment_write_duration_in_seconds",


Please following the Prometheus best practices on naming: https://prometheus.io/docs/practices/naming/

In particular, remove the _in_ part of the metric's name.

EdSchouten · 2025-05-14T11:43:04Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+	cluster := gocql.NewCluster(clusterHosts...)
+	cluster.Keyspace = keyspace
+	cluster.Consistency = writeConsistency
+	if preferredDc != "" {
+		cluster.PoolConfig.HostSelectionPolicy = gocql.TokenAwareHostPolicy(gocql.DCAwareRoundRobinPolicy(preferredDc))
+	}
+
+	if port != 0 {
+		cluster.Port = int(port)
+	}
+
+	if protoVersion != 0 {
+		cluster.ProtoVersion = int(protoVersion)
+	}
+
+	if username != "" {
+		cluster.Authenticator = gocql.PasswordAuthenticator{
+			Username: username,
+			Password: password,
+		}
+	}
+
+	if tlsConfig != nil {
+		cluster.SslOpts = &gocql.SslOptions{
+			Config: tlsConfig,
+		}
+	} else {
+		log.Printf("Not setting up TLS config")
+	}
+
+	if tablePrefix == "" {
+		log.Printf("No table prefix set. Cowardly refusing to proceed.")
+		return blobstore.NewErrorBlobAccess(errors.New("table prefix must be set"))
+	}
+	log.Printf("Table prefix: %s", tablePrefix)
+
+	log.Printf("Creating session in keyspace '%s' for %v", keyspace, clusterHosts)
+
+	session, err := cluster.CreateSession()
+	if err != nil {
+		log.Printf("Unable to create session: %v", err)
+		return blobstore.NewErrorBlobAccess(err)
+	}
+
+	// Create the worker pool for updating the last access times
+	jobs := make(chan job, lastAccessUpdateWorkerCount)
+	for i := 0; i < lastAccessUpdateWorkerCount; i++ {
+		go lastUpdateWorker(jobs)
+	}


As discussed, please move this into a separate function, as it prevents cassandraBlobAccess from being unit tested properly. Also remove the log.Printf() calls, as there is no need to repeat whatever is specified in config. Because once we start to tolerate this, the question becomes: "Why do we print the value of this one configuration option, but not of this other one?" Also get rid of the calls to NewErrorBlobAccess. Any misconfigurations should lead to a fatal startup error.

Extracted the session creation into a helper function and removed the logging.

EdSchouten · 2025-05-14T11:43:27Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+func (a *cassandraBlobAccess) GetCapabilities(ctx context.Context, instanceName bbdigest.InstanceName) (*remoteexecution.ServerCapabilities, error) {
+	return a.capabilitiesProvider.GetCapabilities(ctx, instanceName)
+}


Probably nicer to embed CapabilitiesProvider directly into cassandraBlobAccess, so we don't need this wrapper.

EdSchouten · 2025-05-14T11:52:17Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+		return buffer.NewBufferFromError(err)
+	}
+
+	go a.tables.metadata.updateLastAccessTime(context.Background(), digest, metadata.instanceName, metadata.lastAccess, a.lastAccessUpdateInterval, a.lastAccessUpdate)


Again, even though having no concurrency limits in place might at this point never have been an issue for you, I'd like to see this addressed before this change lands.

EdSchouten · 2025-05-14T11:55:44Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+	g.SetLimit(simultaneousWorkers)
+	for _, digest := range digests.Items() {
+		g.Go(func() error {
+			return a.isMissing(errCtx, digest, &builder, &mu)
+		})
+	}


A downside of using .SetLimit() in combination with a loop like this is that there is no cancelation at all during loop iteration. If a single .isMissing() fails, this code will fail to break immediately, meaning all remaining entries MUST be checked, even if we know they will fail.

Please consider getting rid of the .SetLimit() call and using a separate semaphore, like we do in most other places of the Buildbarn codebase.

How's that?

EdSchouten · 2025-05-14T11:57:16Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+
+func (a *cassandraBlobAccess) getInstanceName(digest bbdigest.Digest) string {
+	instanceName := digest.GetInstanceName().String()
+	if instanceName == a.universalInstanceName {


Please remove support for "universal instance names". I don't think there's any need to do this as part of this backend specifically. Can't it be implemented as a separate decorator for BlobAccess?

Removed. I'll do a follow up PR to set the instance name to a default value if necessary.

EdSchouten · 2025-05-14T11:59:20Z

pkg/blobstore/cassandra/format_test.go

+	longBlob := bytes.Repeat([]byte("This is a very long string!"), 100)
+	query := session.Query("INSERT INTO some_table (id, first_name, long_blob) VALUES (?, ?, ?)", 1234, "Jon Snow", longBlob)
+	expectedOutput := `[query ANY "INSERT INTO some_table (id, first_name, long_blob) VALUES (?, ?, ?)" ` +
+		`[1234 Jon Snow [84 104 105 115 32 105 115 32 97 32 118 101 114 121 32 108 111 110 103 32 115 116 114 105 110 103 33...]]`


Does this mean that if failures occur, this storage backend will potentially dump megabytes of decimal numbers to stderr?

Nope. The formatQuery method caps the length of the output to 100 characters.

EdSchouten · 2025-05-14T11:59:42Z

pkg/proto/configuration/blobstore/blobstore.proto

+
+  // The Cassandra protocol version. This can normally be omitted from
+  // your config.
+  int32 protocolVersion = 4;


protocol_version, preferred_dc.

Missed those. Fixed. Thank you

shs96c

I've landed changes for the simple things. Still need to update to handle the go routines piling up, and I'll get to that ASAP.

shs96c · 2025-05-14T14:40:10Z

pkg/blobstore/cassandra/README.md

+The tables that are required can be set up using command similar to the
+below in a `cqlsh` session:
+
+```shell


Checking the supported languages, I think sql is probably right.

shs96c · 2025-05-14T14:41:42Z

pkg/blobstore/cassandra/format_test.go

+	longBlob := bytes.Repeat([]byte("This is a very long string!"), 100)
+	query := session.Query("INSERT INTO some_table (id, first_name, long_blob) VALUES (?, ?, ?)", 1234, "Jon Snow", longBlob)
+	expectedOutput := `[query ANY "INSERT INTO some_table (id, first_name, long_blob) VALUES (?, ?, ?)" ` +
+		`[1234 Jon Snow [84 104 105 115 32 105 115 32 97 32 118 101 114 121 32 108 111 110 103 32 115 116 114 105 110 103 33...]]`


Nope. The formatQuery method caps the length of the output to 100 characters.

shs96c · 2025-05-14T14:42:23Z

pkg/proto/configuration/blobstore/blobstore.proto

+
+  // The Cassandra protocol version. This can normally be omitted from
+  // your config.
+  int32 protocolVersion = 4;


Missed those. Fixed. Thank you

shs96c · 2025-05-14T14:43:11Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+	"github.com/buildbarn/bb-storage/pkg/blobstore"
+	"github.com/buildbarn/bb-storage/pkg/blobstore/buffer"
+	"github.com/buildbarn/bb-storage/pkg/blobstore/slicing"
+	"github.com/buildbarn/bb-storage/pkg/capabilities"
+	bbdigest "github.com/buildbarn/bb-storage/pkg/digest"
+	"github.com/buildbarn/bb-storage/pkg/util"
+
+	remoteexecution "github.com/bazelbuild/remote-apis/build/bazel/remote/execution/v2"
+	"github.com/gocql/gocql"
+	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"


Is there an automated formatter for this? Or suggested IJ settings?

shs96c · 2025-05-14T14:43:50Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+	"github.com/buildbarn/bb-storage/pkg/blobstore"
+	"github.com/buildbarn/bb-storage/pkg/blobstore/buffer"
+	"github.com/buildbarn/bb-storage/pkg/blobstore/slicing"
+	"github.com/buildbarn/bb-storage/pkg/capabilities"
+	bbdigest "github.com/buildbarn/bb-storage/pkg/digest"
+	"github.com/buildbarn/bb-storage/pkg/util"
+
+	remoteexecution "github.com/bazelbuild/remote-apis/build/bazel/remote/execution/v2"
+	"github.com/gocql/gocql"
+	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"


shs96c · 2025-05-14T15:04:52Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+	totalWritesHist = prometheus.NewHistogram(prometheus.HistogramOpts{
+		Namespace: "buildbarn",
+		Subsystem: "storage",
+		Name:      "cassandra_total_write_duration_in_seconds",


shs96c · 2025-05-14T15:23:09Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+		Namespace: "buildbarn",
+		Subsystem: "storage",
+		Name:      "cassandra_total_write_duration_in_seconds",
+		Help:      "Total amount of time in seconds it takes to write an entire bytestream to Cassandra",


shs96c · 2025-05-14T15:24:52Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+type job struct {
+	task func()
+}


shs96c · 2025-05-14T15:34:43Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+	cluster := gocql.NewCluster(clusterHosts...)
+	cluster.Keyspace = keyspace
+	cluster.Consistency = writeConsistency
+	if preferredDc != "" {
+		cluster.PoolConfig.HostSelectionPolicy = gocql.TokenAwareHostPolicy(gocql.DCAwareRoundRobinPolicy(preferredDc))
+	}
+
+	if port != 0 {
+		cluster.Port = int(port)
+	}
+
+	if protoVersion != 0 {
+		cluster.ProtoVersion = int(protoVersion)
+	}
+
+	if username != "" {
+		cluster.Authenticator = gocql.PasswordAuthenticator{
+			Username: username,
+			Password: password,
+		}
+	}
+
+	if tlsConfig != nil {
+		cluster.SslOpts = &gocql.SslOptions{
+			Config: tlsConfig,
+		}
+	} else {
+		log.Printf("Not setting up TLS config")
+	}
+
+	if tablePrefix == "" {
+		log.Printf("No table prefix set. Cowardly refusing to proceed.")
+		return blobstore.NewErrorBlobAccess(errors.New("table prefix must be set"))
+	}
+	log.Printf("Table prefix: %s", tablePrefix)
+
+	log.Printf("Creating session in keyspace '%s' for %v", keyspace, clusterHosts)
+
+	session, err := cluster.CreateSession()
+	if err != nil {
+		log.Printf("Unable to create session: %v", err)
+		return blobstore.NewErrorBlobAccess(err)
+	}
+
+	// Create the worker pool for updating the last access times
+	jobs := make(chan job, lastAccessUpdateWorkerCount)
+	for i := 0; i < lastAccessUpdateWorkerCount; i++ {
+		go lastUpdateWorker(jobs)
+	}


Extracted the session creation into a helper function and removed the logging.

shs96c · 2025-05-14T16:49:04Z

pkg/blobstore/cassandra/cassandra_blob_access.go

+
+func (a *cassandraBlobAccess) getInstanceName(digest bbdigest.Digest) string {
+	instanceName := digest.GetInstanceName().String()
+	if instanceName == a.universalInstanceName {


Removed. I'll do a follow up PR to set the instance name to a default value if necessary.

shs96c added 3 commits May 6, 2025 14:37

Add a cassandra-backed storage option

f453e1f

Also suggest timeouts for other tables

47b945a

Make sure the test compiles

2698fe1

EdSchouten reviewed May 14, 2025

View reviewed changes

Respond to review comments

9a2f628

shs96c commented May 14, 2025

View reviewed changes

shs96c added 2 commits May 14, 2025 19:57

Use a non-blocking put on a channel to avoid having too many go routines

d91fca0

Use a semaphore rather than an limit count for the errgroup

bb26592

Add a cassandra-backed storage option #246

Are you sure you want to change the base?

Add a cassandra-backed storage option #246

Uh oh!

Conversation

shs96c commented May 6, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shs96c left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants