Use bytes instead of UUIDs for persistence interfaces

Author: bergundy

common/membership/ringpop/monitor.go

@@ -64,7 +64,7 @@ type monitor struct {
 	logger                    log.Logger
 	metadataManager           persistence.ClusterMetadataManager
 	broadcastHostPortResolver func() (string, error)
-	hostID                    uuid.UUID
+	hostID                    []byte
 	initialized               *future.FutureImpl[struct{}]
 }
 
@@ -87,6 +87,8 @@ func newMonitor(
 		lifecycleCtx,
 		headers.SystemBackgroundHighCallerInfo,
 	)
+	hostID, _ := uuid.New().MarshalBinary()
+	// MarshalBinary should never error.
 
 	rpo := &monitor{
 		status: common.DaemonStatusInitialized,
@@ -101,7 +103,7 @@ func newMonitor(
 		logger:                    logger,
 		metadataManager:           metadataManager,
 		broadcastHostPortResolver: broadcastHostPortResolver,
-		hostID:                    uuid.New(),
+		hostID:                    hostID,
 		initialized:               future.NewFuture[struct{}](),
 		maxJoinDuration:           maxJoinDuration,
 		propagationTime:           propagationTime,
@@ -251,10 +253,14 @@ func (rpo *monitor) upsertMyMembership(
 	err := rpo.metadataManager.UpsertClusterMembership(ctx, request)
 
 	if err == nil {
+		hostID, err := uuid.FromBytes(request.HostID)
+		if err != nil {
+			return err
+		}
 		rpo.logger.Debug("Membership heartbeat upserted successfully",
 			tag.Address(request.RPCAddress.String()),
 			tag.Port(int(request.RPCPort)),
-			tag.HostID(request.HostID.String()))
+			tag.HostID(hostID.String()))
 	}
 
 	return err
@@ -313,10 +319,14 @@ func (rpo *monitor) startHeartbeat(broadcastHostport string) error {
 	// read side by filtering on the last time a heartbeat was seen.
 	err = rpo.upsertMyMembership(rpo.lifecycleCtx, req)
 	if err == nil {
+		hostID, err := uuid.FromBytes(rpo.hostID)
+		if err != nil {
+			return err
+		}
 		rpo.logger.Info("Membership heartbeat upserted successfully",
 			tag.Address(broadcastAddress.String()),
 			tag.Port(int(broadcastPort)),
-			tag.HostID(rpo.hostID.String()))
+			tag.HostID(hostID.String()))
 
 		rpo.startHeartbeatUpsertLoop(req)
 	}
@@ -350,7 +360,7 @@ func (rpo *monitor) fetchCurrentBootstrapHostports() ([]string, error) {
 		nextPageToken = resp.NextPageToken
 
 		// Stop iterating once we have either 500 unique ip:port combos or there is no more results.
-		if nextPageToken == nil || len(set) >= 500 {
+		if len(nextPageToken) == 0 || len(set) >= 500 {
 			bootstrapHostPorts := make([]string, 0, len(set))
 			for k := range set {
 				bootstrapHostPorts = append(bootstrapHostPorts, k)

common/membership/ringpop/test_cluster.go

@@ -91,8 +91,11 @@ func newTestCluster(
 		logger.Error("unable to split host port", tag.Error(err))
 		return nil
 	}
+	// MarshalBinary never fails for UUIDs
+	hostID, _ := uuid.New().MarshalBinary()
+
 	seedMember := &persistence.ClusterMember{
-		HostID:        uuid.New(),
+		HostID:        hostID,
 		RPCAddress:    seedAddress,
 		RPCPort:       seedPort,
 		SessionStart:  time.Now().UTC(),
@@ -104,12 +107,13 @@ func newTestCluster(
 		func(_ context.Context, _ *persistence.GetClusterMembersRequest) (*persistence.GetClusterMembersResponse, error) {
 			res := &persistence.GetClusterMembersResponse{ActiveMembers: []*persistence.ClusterMember{seedMember}}
 
+			hostID, _ := uuid.New().MarshalBinary()
 			if firstGetClusterMemberCall {
 				// The first time GetClusterMembers is invoked, we simulate returning a stale/bad heartbeat.
 				// All subsequent calls only return the single "good" seed member
 				// This ensures that we exercise the retry path in bootstrap properly.
 				badSeedMember := &persistence.ClusterMember{
-					HostID:        uuid.New(),
+					HostID:        hostID,
 					RPCAddress:    seedAddress,
 					RPCPort:       seedPort + 1,
 					SessionStart:  time.Now().UTC(),

common/persistence/cassandra/cluster_metadata_store.go

@@ -2,12 +2,10 @@ package cassandra
 
 import (
 	"context"
-	"fmt"
 	"net"
 	"strings"
 	"time"
 
-	"github.com/google/uuid"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/common/log"
 	p "go.temporal.io/server/common/persistence"
@@ -177,13 +175,9 @@ func (m *ClusterMetadataStore) GetClusterMembers(
 	queryString.WriteString(templateGetClusterMembership)
 	operands = append(operands, constMembershipPartition)
 
-	if request.HostIDEquals != uuid.Nil {
+	if len(request.HostIDEquals) != 0 {
 		queryString.WriteString(templateWithHostIDSuffix)
-		hostIDEqualsBytes, err := request.HostIDEquals.MarshalBinary()
-		if err != nil {
-			return nil, gocql.ConvertError("GetClusterMembers", fmt.Errorf("failed to marshal HostIDEquals: %w", err))
-		}
-		operands = append(operands, hostIDEqualsBytes)
+		operands = append(operands, request.HostIDEquals)
 	}
 
 	if request.RPCAddressEquals != nil {
@@ -225,7 +219,7 @@ func (m *ClusterMetadataStore) GetClusterMembers(
 
 	for iter.Scan(&cqlHostID, &rpcAddress, &rpcPort, &role, &sessionStart, &lastHeartbeat, &cassNow, &ttl) {
 		member := p.ClusterMember{
-			HostID:        uuid.UUID(cqlHostID),
+			HostID:        cqlHostID,
 			RPCAddress:    rpcAddress,
 			RPCPort:       rpcPort,
 			Role:          role,
@@ -253,22 +247,18 @@ func (m *ClusterMetadataStore) UpsertClusterMembership(
 	ctx context.Context,
 	request *p.UpsertClusterMembershipRequest,
 ) error {
-	hostIDBytes, err := request.HostID.MarshalBinary()
-	if err != nil {
-		return gocql.ConvertError("UpsertClusterMembership", fmt.Errorf("failed to marshal HostID: %w", err))
-	}
 	query := m.session.Query(
 		templateUpsertActiveClusterMembership,
 		constMembershipPartition,
-		hostIDBytes,
+		request.HostID,
 		request.RPCAddress,
 		request.RPCPort,
 		request.Role,
 		request.SessionStart,
 		time.Now().UTC(),
 		int64(request.RecordExpiry.Seconds()),
 	).WithContext(ctx)
-	err = query.Exec()
+	err := query.Exec()
 
 	if err != nil {
 		return gocql.ConvertError("UpsertClusterMembership", err)

common/persistence/data_interfaces.go

@@ -9,7 +9,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/uuid"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	historypb "go.temporal.io/api/history/v1"
@@ -995,7 +994,7 @@ type (
 	GetClusterMembersRequest struct {
 		LastHeartbeatWithin time.Duration
 		RPCAddressEquals    net.IP
-		HostIDEquals        uuid.UUID
+		HostIDEquals        []byte
 		RoleEquals          ServiceType
 		SessionStartedAfter time.Time
 		NextPageToken       []byte
@@ -1011,7 +1010,7 @@ type (
 	// ClusterMember is used as a response to GetClusterMembers
 	ClusterMember struct {
 		Role          ServiceType
-		HostID        uuid.UUID
+		HostID        []byte
 		RPCAddress    net.IP
 		RPCPort       uint16
 		SessionStart  time.Time
@@ -1022,7 +1021,7 @@ type (
 	// UpsertClusterMembershipRequest is the request to UpsertClusterMembership
 	UpsertClusterMembershipRequest struct {
 		Role         ServiceType
-		HostID       uuid.UUID
+		HostID       []byte
 		RPCAddress   net.IP
 		RPCPort      uint16
 		SessionStart time.Time

common/persistence/persistence-tests/cluster_metadata_manager.go

@@ -61,17 +61,18 @@ func (s *ClusterMetadataManagerSuite) TestClusterMembershipEmptyInitially() {
 
 // TestClusterMembershipUpsertCanReadAny verifies that we can UpsertClusterMembership and read our result
 func (s *ClusterMetadataManagerSuite) TestClusterMembershipUpsertCanReadAny() {
-
+	hostID, err := uuid.New().MarshalBinary()
+	s.NoError(err)
 	req := &p.UpsertClusterMembershipRequest{
-		HostID:       uuid.New(),
+		HostID:       hostID,
 		RPCAddress:   net.ParseIP("127.0.0.2"),
 		RPCPort:      123,
 		Role:         p.Frontend,
 		SessionStart: time.Now().UTC(),
 		RecordExpiry: time.Second,
 	}
 
-	err := s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
+	err = s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
 	s.Nil(err)
 
 	resp, err := s.ClusterMetadataManager.GetClusterMembers(s.ctx, &p.GetClusterMembersRequest{})
@@ -90,17 +91,19 @@ func (s *ClusterMetadataManagerSuite) TestClusterMembershipUpsertCanPageRead() {
 		hostID := uuid.New()
 
 		expectedIds[hostID.String()]++
+		hostIDBytes, err := hostID.MarshalBinary()
+		s.NoError(err)
 
 		req := &p.UpsertClusterMembershipRequest{
-			HostID:       hostID,
+			HostID:       hostIDBytes,
 			RPCAddress:   net.ParseIP("127.0.0.2"),
 			RPCPort:      123,
 			Role:         p.Frontend,
 			SessionStart: time.Now().UTC(),
 			RecordExpiry: 3 * time.Second,
 		}
 
-		err := s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
+		err = s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
 		s.NoError(err)
 	}
 
@@ -111,7 +114,9 @@ func (s *ClusterMetadataManagerSuite) TestClusterMembershipUpsertCanPageRead() {
 		s.NoError(err)
 		nextPageToken = resp.NextPageToken
 		for _, member := range resp.ActiveMembers {
-			expectedIds[member.HostID.String()]--
+			hostID, err := uuid.FromBytes(member.HostID)
+			s.NoError(err)
+			expectedIds[hostID.String()]--
 			hostCount++
 		}
 
@@ -145,16 +150,18 @@ func (s *ClusterMetadataManagerSuite) validateUpsert(req *p.UpsertClusterMembers
 // TestClusterMembershipReadFiltersCorrectly verifies that we can UpsertClusterMembership and read our result using filters
 func (s *ClusterMetadataManagerSuite) TestClusterMembershipReadFiltersCorrectly() {
 	now := time.Now().UTC()
+	hostID, err := uuid.New().MarshalBinary()
+	s.NoError(err)
 	req := &p.UpsertClusterMembershipRequest{
-		HostID:       uuid.New(),
+		HostID:       hostID,
 		RPCAddress:   net.ParseIP("127.0.0.2"),
 		RPCPort:      123,
 		Role:         p.Frontend,
 		SessionStart: now,
 		RecordExpiry: time.Second * 4,
 	}
 
-	err := s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
+	err = s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
 	s.Nil(err)
 
 	resp, err := s.ClusterMetadataManager.GetClusterMembers(
@@ -203,16 +210,18 @@ func (s *ClusterMetadataManagerSuite) TestClusterMembershipReadFiltersCorrectly(
 
 // TestClusterMembershipUpsertExpiresCorrectly verifies RecordExpiry functions properly for ClusterMembership records
 func (s *ClusterMetadataManagerSuite) TestClusterMembershipUpsertExpiresCorrectly() {
+	hostID, err := uuid.New().MarshalBinary()
+	s.NoError(err)
 	req := &p.UpsertClusterMembershipRequest{
-		HostID:       uuid.New(),
+		HostID:       hostID,
 		RPCAddress:   net.ParseIP("127.0.0.2"),
 		RPCPort:      123,
 		Role:         p.Frontend,
 		SessionStart: time.Now().UTC(),
 		RecordExpiry: time.Second,
 	}
 
-	err := s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
+	err = s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
 	s.NoError(err)
 
 	err = s.ClusterMetadataManager.PruneClusterMembership(s.ctx, &p.PruneClusterMembershipRequest{MaxRecordsPruned: 100})
@@ -260,16 +269,18 @@ func (s *ClusterMetadataManagerSuite) waitForPrune(waitFor time.Duration) {
 
 // TestClusterMembershipUpsertInvalidExpiry verifies we cannot specify a non-positive RecordExpiry duration
 func (s *ClusterMetadataManagerSuite) TestClusterMembershipUpsertInvalidExpiry() {
+	hostID, err := uuid.New().MarshalBinary()
+	s.NoError(err)
 	req := &p.UpsertClusterMembershipRequest{
-		HostID:       uuid.New(),
+		HostID:       hostID,
 		RPCAddress:   net.ParseIP("127.0.0.2"),
 		RPCPort:      123,
 		Role:         p.Frontend,
 		SessionStart: time.Now().UTC(),
 		RecordExpiry: time.Second * 0,
 	}
 
-	err := s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
+	err = s.ClusterMetadataManager.UpsertClusterMembership(s.ctx, req)
 	s.NotNil(err)
 	s.IsType(err, p.ErrInvalidMembershipExpiry)
 }

common/persistence/sql/cluster_metadata.go

@@ -6,7 +6,6 @@ import (
 	"net"
 	"time"
 
-	"github.com/google/uuid"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/common/log"
 	p "go.temporal.io/server/common/persistence"
@@ -136,14 +135,9 @@ func (s *sqlClusterMetadataManager) GetClusterMembers(
 		return nil, serviceerror.NewInternal("page token is corrupted.")
 	}
 
-	hostIDEqualsBytes, err := request.HostIDEquals.MarshalBinary()
-	if err != nil {
-		return nil, serviceerror.NewInvalidArgumentf("unable to marshal HostIDEquals: %v", err)
-	}
-
 	now := time.Now().UTC()
 	filter := &sqlplugin.ClusterMembershipFilter{
-		HostIDEquals:        hostIDEqualsBytes,
+		HostIDEquals:        request.HostIDEquals,
 		RoleEquals:          request.RoleEquals,
 		RecordExpiryAfter:   now,
 		SessionStartedAfter: request.SessionStartedAfter,
@@ -170,13 +164,8 @@ func (s *sqlClusterMetadataManager) GetClusterMembers(
 
 	convertedRows := make([]*p.ClusterMember, 0, len(rows))
 	for _, row := range rows {
-		hostIDUuid, err := uuid.ParseBytes(row.HostID)
-		if err != nil {
-			return nil, serviceerror.NewInternalf("unable to parse HostID: %v", err)
-		}
-
 		convertedRows = append(convertedRows, &p.ClusterMember{
-			HostID:        hostIDUuid,
+			HostID:        row.HostID,
 			Role:          row.Role,
 			RPCAddress:    net.ParseIP(row.RPCAddress),
 			RPCPort:       row.RPCPort,
@@ -201,14 +190,9 @@ func (s *sqlClusterMetadataManager) UpsertClusterMembership(
 ) error {
 	now := time.Now().UTC()
 	recordExpiry := now.Add(request.RecordExpiry)
-	hostIDBytes, err := request.HostID.MarshalBinary()
-	if err != nil {
-		return serviceerror.NewInvalidArgumentf("unable to marshal HostID: %v", err)
-	}
-
-	_, err = s.DB.UpsertClusterMembership(ctx, &sqlplugin.ClusterMembershipRow{
+	_, err := s.DB.UpsertClusterMembership(ctx, &sqlplugin.ClusterMembershipRow{
 		Role:          request.Role,
-		HostID:        hostIDBytes,
+		HostID:        request.HostID,
 		RPCAddress:    request.RPCAddress.String(),
 		RPCPort:       request.RPCPort,
 		SessionStart:  request.SessionStart,

service/frontend/admin_handler.go

@@ -1134,11 +1134,15 @@ func (adh *AdminHandler) ListClusterMembers(
 	if err != nil {
 		return nil, serviceerror.NewInvalidArgumentf("host ID %q is not a valid UUID: %v", request.GetHostId(), err)
 	}
+	hostIDEqualBytes, err := hostIDEqual.MarshalBinary()
+	if err != nil {
+		return nil, serviceerror.NewInternalf("unable to marshal host ID %q to bytes: %v", request.GetHostId(), err)
+	}
 
 	resp, err := metadataMgr.GetClusterMembers(ctx, &persistence.GetClusterMembersRequest{
 		LastHeartbeatWithin: heartbit,
 		RPCAddressEquals:    net.ParseIP(request.GetRpcAddress()),
-		HostIDEquals:        hostIDEqual,
+		HostIDEquals:        hostIDEqualBytes,
 		RoleEquals:          persistence.ServiceType(request.GetRole()),
 		SessionStartedAfter: startedTime,
 		PageSize:            int(request.GetPageSize()),
@@ -1150,9 +1154,13 @@ func (adh *AdminHandler) ListClusterMembers(
 
 	var activeMembers []*clusterspb.ClusterMember
 	for _, member := range resp.ActiveMembers {
+		u, err := uuid.FromBytes(member.HostID)
+		if err != nil {
+			return nil, serviceerror.NewInternalf("unable to parse host ID bytes to UUID: %v", err)
+		}
 		activeMembers = append(activeMembers, &clusterspb.ClusterMember{
 			Role:             enumsspb.ClusterMemberRole(member.Role),
-			HostId:           member.HostID.String(),
+			HostId:           u.String(),
 			RpcAddress:       member.RPCAddress.String(),
 			RpcPort:          int32(member.RPCPort),
 			SessionStartTime: timestamppb.New(member.SessionStart),