rename types to etcdtypes

Signed-off-by: Seva Kaloshin <[email protected]>

Author: arzonus

service/sharddistributor/store/etcd/etcdkeys/etcdkeys.go

@@ -3,7 +3,6 @@ package etcdkeys
 import (
 	"fmt"
 	"strings"
-	"time"
 )
 
 const (
@@ -42,8 +41,6 @@ func BuildExecutorPrefix(prefix string, namespace string) string {
 }
 
 func BuildExecutorKey(prefix string, namespace, executorID, keyType string) (string, error) {
-
-	time.Now().Unix()
 	// We allow an empty key, to build the full prefix
 	if !isValidKeyType(keyType) && keyType != "" {
 		return "", fmt.Errorf("invalid key type: %s", keyType)

service/sharddistributor/store/etcd/executorstore/types/state.go renamed to service/sharddistributor/store/etcd/etcdtypes/state.go

@@ -1,4 +1,4 @@
-package types
+package etcdtypes
 
 import (
 	"github.com/uber/cadence/common/types"

service/sharddistributor/store/etcd/executorstore/types/time.go renamed to service/sharddistributor/store/etcd/etcdtypes/time.go

@@ -1,4 +1,4 @@
-package types
+package etcdtypes
 
 import "time"
 

service/sharddistributor/store/etcd/executorstore/etcdstore.go

@@ -10,8 +10,6 @@ import (
 	"fmt"
 	"time"
 
-	types2 "github.com/uber/cadence/service/sharddistributor/store/etcd/executorstore/types"
-
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/fx"
 
@@ -22,6 +20,7 @@ import (
 	"github.com/uber/cadence/service/sharddistributor/config"
 	"github.com/uber/cadence/service/sharddistributor/store"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/etcdkeys"
+	"github.com/uber/cadence/service/sharddistributor/store/etcd/etcdtypes"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/executorstore/common"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/executorstore/shardcache"
 )
@@ -43,8 +42,8 @@ type executorStoreImpl struct {
 type shardStatisticsUpdate struct {
 	key             string
 	shardID         string
-	stats           types2.ShardStatistics
-	desiredLastMove types2.Time // intended LastMoveTime for this update
+	stats           etcdtypes.ShardStatistics
+	desiredLastMove etcdtypes.Time // intended LastMoveTime for this update
 }
 
 // ExecutorStoreParams defines the dependencies for the etcd store, for use with fx.
@@ -139,7 +138,7 @@ func (s *executorStoreImpl) RecordHeartbeat(ctx context.Context, namespace, exec
 
 	// Build all operations including metadata
 	ops := []clientv3.Op{
-		clientv3.OpPut(heartbeatETCDKey, types2.FromTime(request.LastHeartbeat)),
+		clientv3.OpPut(heartbeatETCDKey, etcdtypes.FromTime(request.LastHeartbeat)),
 		clientv3.OpPut(stateETCDKey, string(jsonState)),
 		clientv3.OpPut(reportedShardsETCDKey, string(reportedShardsData)),
 	}
@@ -174,7 +173,7 @@ func (s *executorStoreImpl) GetHeartbeat(ctx context.Context, namespace string,
 	}
 
 	heartbeatState := &store.HeartbeatState{}
-	assignedState := &types2.AssignedState{}
+	assignedState := &etcdtypes.AssignedState{}
 	found := false
 
 	for _, kv := range resp.Kvs {
@@ -188,7 +187,7 @@ func (s *executorStoreImpl) GetHeartbeat(ctx context.Context, namespace string,
 		found = true // We found at least one valid key part for the executor.
 		switch keyType {
 		case etcdkeys.ExecutorHeartbeatKey:
-			heartbeatState.LastHeartbeat, err = types2.ToTime(value)
+			heartbeatState.LastHeartbeat, err = etcdtypes.ToTime(value)
 			if err != nil {
 				return nil, nil, fmt.Errorf("parse heartbeat timestamp: %w", err)
 			}
@@ -240,7 +239,7 @@ func (s *executorStoreImpl) GetState(ctx context.Context, namespace string) (*st
 
 		switch keyType {
 		case etcdkeys.ExecutorHeartbeatKey:
-			heartbeat.LastHeartbeat, err = types2.ToTime(value)
+			heartbeat.LastHeartbeat, err = etcdtypes.ToTime(value)
 			if err != nil {
 				return nil, fmt.Errorf("parse heartbeat timestamp: %w", err)
 			}
@@ -253,7 +252,7 @@ func (s *executorStoreImpl) GetState(ctx context.Context, namespace string) (*st
 				return nil, fmt.Errorf("parse reported shards: %w", err)
 			}
 		case etcdkeys.ExecutorAssignedStateKey:
-			var assignedRaw types2.AssignedState
+			var assignedRaw etcdtypes.AssignedState
 			if err := common.DecompressAndUnmarshal(kv.Value, &assignedRaw); err != nil {
 				return nil, fmt.Errorf("parse assigned shards: %w, %s", err, value)
 			}
@@ -278,7 +277,7 @@ func (s *executorStoreImpl) GetState(ctx context.Context, namespace string) (*st
 		if shardKeyType != etcdkeys.ShardStatisticsKey {
 			continue
 		}
-		var shardStatistic types2.ShardStatistics
+		var shardStatistic etcdtypes.ShardStatistics
 		if err := common.DecompressAndUnmarshal(kv.Value, &shardStatistic); err != nil {
 			continue
 		}
@@ -355,7 +354,7 @@ func (s *executorStoreImpl) AssignShards(ctx context.Context, namespace string,
 		if err != nil {
 			return fmt.Errorf("build executor assigned state key: %w", err)
 		}
-		value, err := json.Marshal(types2.FromAssignedState(&state))
+		value, err := json.Marshal(etcdtypes.FromAssignedState(&state))
 		if err != nil {
 			return fmt.Errorf("marshal assigned shards for executor %s: %w", executorID, err)
 		}
@@ -438,8 +437,8 @@ func (s *executorStoreImpl) AssignShard(ctx context.Context, namespace, shardID,
 			return fmt.Errorf("get executor assigned state: %w", err)
 		}
 
-		var state types2.AssignedState
-		var shardStats types2.ShardStatistics
+		var state etcdtypes.AssignedState
+		var shardStats etcdtypes.ShardStatistics
 		modRevision := int64(0) // A revision of 0 means the key doesn't exist yet.
 
 		if len(resp.Kvs) > 0 {
@@ -469,12 +468,12 @@ func (s *executorStoreImpl) AssignShard(ctx context.Context, namespace, shardID,
 			// Statistics already exist, update the last move time.
 			// This can happen if the shard was previously assigned to an executor, and a lookup happens after the executor is deleted,
 			// AssignShard is then called to assign the shard to a new executor.
-			shardStats.LastMoveTime = types2.Time(now)
+			shardStats.LastMoveTime = etcdtypes.Time(now)
 		} else {
 			// Statistics don't exist, initialize them.
 			shardStats.SmoothedLoad = 0
-			shardStats.LastUpdateTime = types2.Time(now)
-			shardStats.LastMoveTime = types2.Time(now)
+			shardStats.LastUpdateTime = etcdtypes.Time(now)
+			shardStats.LastMoveTime = etcdtypes.Time(now)
 		}
 
 		// 2. Get the executor state.
@@ -677,22 +676,22 @@ func (s *executorStoreImpl) prepareShardStatisticsUpdates(ctx context.Context, n
 				return nil, fmt.Errorf("get shard statistics: %w", err)
 			}
 
-			stats := types2.ShardStatistics{}
+			stats := etcdtypes.ShardStatistics{}
 
 			if len(statsResp.Kvs) > 0 {
 				if err := common.DecompressAndUnmarshal(statsResp.Kvs[0].Value, &stats); err != nil {
 					return nil, fmt.Errorf("parse shard statistics: %w", err)
 				}
 			} else {
 				stats.SmoothedLoad = 0
-				stats.LastUpdateTime = types2.Time(now)
+				stats.LastUpdateTime = etcdtypes.Time(now)
 			}
 
 			updates = append(updates, shardStatisticsUpdate{
 				key:             shardStatisticsKey,
 				shardID:         shardID,
 				stats:           stats,
-				desiredLastMove: types2.Time(now),
+				desiredLastMove: etcdtypes.Time(now),
 			})
 		}
 	}

service/sharddistributor/store/etcd/executorstore/etcdstore_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	types2 "github.com/uber/cadence/service/sharddistributor/store/etcd/executorstore/types"
+	types3 "github.com/uber/cadence/service/sharddistributor/store/etcd/types"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -60,7 +60,7 @@ func TestRecordHeartbeat(t *testing.T) {
 	resp, err := tc.Client.Get(ctx, heartbeatKey)
 	require.NoError(t, err)
 	assert.Equal(t, int64(1), resp.Count, "Heartbeat key should exist")
-	assert.Equal(t, types2.FromTime(now), string(resp.Kvs[0].Value))
+	assert.Equal(t, types3.FromTime(now), string(resp.Kvs[0].Value))
 
 	resp, err = tc.Client.Get(ctx, stateKey)
 	require.NoError(t, err)
@@ -545,7 +545,7 @@ func TestShardStatisticsPersistence(t *testing.T) {
 	stats := store.ShardStatistics{SmoothedLoad: 12.5, LastUpdateTime: time.Unix(1234, 0).UTC(), LastMoveTime: time.Unix(5678, 0).UTC()}
 	shardStatsKey, err := etcdkeys.BuildShardKey(tc.EtcdPrefix, tc.Namespace, shardID, etcdkeys.ShardStatisticsKey)
 	require.NoError(t, err)
-	payload, err := json.Marshal(types2.FromShardStatistics(&stats))
+	payload, err := json.Marshal(types3.FromShardStatistics(&stats))
 	require.NoError(t, err)
 	_, err = tc.Client.Put(ctx, shardStatsKey, string(payload))
 	require.NoError(t, err)

service/sharddistributor/store/etcd/executorstore/shardcache/namespaceshardcache.go

@@ -10,9 +10,8 @@ import (
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/service/sharddistributor/store"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/etcdkeys"
+	"github.com/uber/cadence/service/sharddistributor/store/etcd/etcdtypes"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/executorstore/common"
-	"github.com/uber/cadence/service/sharddistributor/store/etcd/executorstore/types"
-
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
@@ -176,7 +175,7 @@ func (n *namespaceShardToExecutor) refreshExecutorState(ctx context.Context) err
 		case etcdkeys.ExecutorAssignedStateKey:
 			shardOwner := getOrCreateShardOwner(shardOwners, executorID)
 
-			var assignedState types.AssignedState
+			var assignedState etcdtypes.AssignedState
 			err = common.DecompressAndUnmarshal(kv.Value, &assignedState)
 			if err != nil {
 				return fmt.Errorf("parse assigned state: %w", err)

service/sharddistributor/store/etcd/executorstore/shardcache/namespaceshardcache_test.go

@@ -12,15 +12,15 @@ import (
 
 	"github.com/uber/cadence/common/log/testlogger"
 	"github.com/uber/cadence/common/types"
-	"github.com/uber/cadence/service/sharddistributor/store"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/etcdkeys"
+	"github.com/uber/cadence/service/sharddistributor/store/etcd/etcdtypes"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/testhelper"
 )
 
 // setupExecutorWithShards creates an executor in etcd with assigned shards and metadata
 func setupExecutorWithShards(t *testing.T, testCluster *testhelper.StoreTestCluster, namespace, executorID string, shards []string, metadata map[string]string) {
 	// Create assigned state
-	assignedState := &store.AssignedState{
+	assignedState := &etcdtypes.AssignedState{
 		AssignedShards: make(map[string]*types.ShardAssignment),
 	}
 	for _, shardID := range shards {