Upgrade golang 1.17 (#98)

* upgrade to golang 1.17

Signed-off-by: Fabiano Graças <[email protected]>

# Conflicts:
#	go.mod
#	go.sum

* improve after shell lint

Signed-off-by: Fabiano Graças <[email protected]>

* improve after upgrade docker image (used by the build system)

Signed-off-by: Fabiano Graças <[email protected]>

* remove not needed variable

Signed-off-by: Fabiano Graças <[email protected]>

* apply fixes after security scan (hmake test)

Signed-off-by: Fabiano Graças <[email protected]>

* add missing package after merge with latest master branch code.

Signed-off-by: Fabiano Graças <[email protected]>

* improve docker layer

Signed-off-by: Fabiano Graças <[email protected]>

* upgrade packages

Signed-off-by: Fabiano Graças <[email protected]>

Co-authored-by: Fabiano Graças <[email protected]>

Author: fafg

.gitignore

@@ -15,3 +15,5 @@
 .vscode
 *_mock_test.go
 filenames
+
+.DS_Store

HyperMake

@@ -21,7 +21,6 @@ targets:
     watches:
       - go.mod
     cmds:
-      - export GO111MODULE=on
       - go mod download
       - go mod vendor
       - go mod tidy
@@ -88,5 +87,5 @@ settings:
   default-targets:
     - test
   docker:
-    image: 'vmware/go-kcl-toolchain:0.1.3'
+    image: 'vmware/go-kcl-toolchain:0.1.4'
     src-volume: /go/src/github.com/vmware/vmware-go-kcl

clientlibrary/utils/random.go

@@ -16,10 +16,13 @@
  * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
  * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  */
+
+// Package utils
 package utils
 
 import (
-	"math/rand"
+	"crypto/rand"
+	"math/big"
 	"time"
 )
 
@@ -32,11 +35,13 @@ const (
 
 func RandStringBytesMaskImpr(n int) string {
 	b := make([]byte, n)
-	rand.Seed(time.Now().UTC().UnixNano())
-	// A rand.Int63() generates 63 random bits, enough for letterIdxMax letters!
-	for i, cache, remain := n-1, rand.Int63(), letterIdxMax; i >= 0; {
+	seed := time.Now().UTC().UnixNano()
+	rnd, _ := rand.Int(rand.Reader, big.NewInt(seed))
+	// A rand.Int64() generates 64 random bits, enough for letterIdxMax letters!
+	for i, cache, remain := n-1, rnd.Int64(), letterIdxMax; i >= 0; {
 		if remain == 0 {
-			cache, remain = rand.Int63(), letterIdxMax
+			rnd, _ = rand.Int(rand.Reader, big.NewInt(seed))
+			cache, remain = rnd.Int64(), letterIdxMax
 		}
 		if idx := int(cache & letterIdxMask); idx < len(letterBytes) {
 			b[i] = letterBytes[idx]

clientlibrary/utils/random_test.go

@@ -20,7 +20,6 @@ package utils
 
 import (
 	"fmt"
-	"math/rand"
 	"testing"
 	"time"
 )
@@ -32,17 +31,18 @@ func TestRandom(t *testing.T) {
 		if s1 == s2 {
 			t.Fatalf("failed in generating random string. s1: %s, s2: %s", s1, s2)
 		}
+		fmt.Println(s1)
+		fmt.Println(s2)
 	}
 }
 
 func TestRandomNum(t *testing.T) {
-	rand.Seed(time.Now().UTC().UnixNano())
-
 	for i := 0; i < 10; i++ {
-		s1 := rand.Int63()
-		s2 := rand.Int63()
+		seed := time.Now().UTC().Second()
+		s1 := RandStringBytesMaskImpr(seed)
+		s2 := RandStringBytesMaskImpr(seed)
 		if s1 == s2 {
-			t.Fatalf("failed in generating random string. s1: %d, s2: %d", s1, s2)
+			t.Fatalf("failed in generating random string. s1: %s, s2: %s", s1, s2)
 		}
 		fmt.Println(s1)
 		fmt.Println(s2)

clientlibrary/worker/worker.go

@@ -16,6 +16,8 @@
  * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
  * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  */
+
+// Package worker
 // The implementation is derived from https://github.com/patrobinson/gokini
 //
 // Copyright 2018 Patrick robinson
@@ -28,8 +30,9 @@
 package worker
 
 import (
+	"crypto/rand"
 	"errors"
-	"math/rand"
+	"math/big"
 	"sync"
 	"time"
 
@@ -45,11 +48,9 @@ import (
 	par "github.com/vmware/vmware-go-kcl/clientlibrary/partition"
 )
 
-/**
- * Worker is the high level class that Kinesis applications use to start processing data. It initializes and oversees
- * different components (e.g. syncing shard and lease information, tracking shard assignments, and processing data from
- * the shards).
- */
+//Worker is the high level class that Kinesis applications use to start processing data. It initializes and oversees
+//different components (e.g. syncing shard and lease information, tracking shard assignments, and processing data from
+//the shards).
 type Worker struct {
 	streamName  string
 	regionName  string
@@ -66,7 +67,7 @@ type Worker struct {
 	waitGroup *sync.WaitGroup
 	done      bool
 
-	rng *rand.Rand
+	randomSeed int64
 
 	shardStatus          map[string]*par.ShardStatus
 	shardStealInProgress bool
@@ -80,9 +81,6 @@ func NewWorker(factory kcl.IRecordProcessorFactory, kclConfig *config.KinesisCli
 		mService = metrics.NoopMonitoringService{}
 	}
 
-	// Create a pseudo-random number generator and seed it.
-	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
-
 	return &Worker{
 		streamName:       kclConfig.StreamName,
 		regionName:       kclConfig.RegionName,
@@ -91,7 +89,7 @@ func NewWorker(factory kcl.IRecordProcessorFactory, kclConfig *config.KinesisCli
 		kclConfig:        kclConfig,
 		mService:         mService,
 		done:             false,
-		rng:              rng,
+		randomSeed:       time.Now().UTC().UnixNano(),
 	}
 }
 
@@ -108,7 +106,7 @@ func (w *Worker) WithCheckpointer(checker chk.Checkpointer) *Worker {
 	return w
 }
 
-// Run starts consuming data from the stream, and pass it to the application record processors.
+// Start Run starts consuming data from the stream, and pass it to the application record processors.
 func (w *Worker) Start() error {
 	log := w.kclConfig.Logger
 	if err := w.initialize(); err != nil {
@@ -133,7 +131,7 @@ func (w *Worker) Start() error {
 	return nil
 }
 
-// Shutdown signals worker to shutdown. Worker will try initiating shutdown of all record processors.
+// Shutdown signals worker to shut down. Worker will try initiating shutdown of all record processors.
 func (w *Worker) Shutdown() {
 	log := w.kclConfig.Logger
 	log.Infof("Worker shutdown in requested.")
@@ -258,7 +256,8 @@ func (w *Worker) eventLoop() {
 		// starts at the same time, this decreases the probability of them calling
 		// kinesis.DescribeStream at the same time, and hit the hard-limit on aws API calls.
 		// On average the period remains the same so that doesn't affect behavior.
-		shardSyncSleep := w.kclConfig.ShardSyncIntervalMillis/2 + w.rng.Intn(w.kclConfig.ShardSyncIntervalMillis)
+		rnd, _ := rand.Int(rand.Reader, big.NewInt(int64(w.kclConfig.ShardSyncIntervalMillis)))
+		shardSyncSleep := w.kclConfig.ShardSyncIntervalMillis/2 + int(rnd.Int64())
 
 		err := w.syncShard()
 		if err != nil {
@@ -290,7 +289,7 @@ func (w *Worker) eventLoop() {
 
 				err := w.checkpointer.FetchCheckpoint(shard)
 				if err != nil {
-					// checkpoint may not existed yet is not an error condition.
+					// checkpoint may not exist yet is not an error condition.
 					if err != chk.ErrSequenceIDNotFound {
 						log.Warnf("Couldn't fetch checkpoint: %+v", err)
 						// move on to next shard
@@ -371,7 +370,7 @@ func (w *Worker) rebalance() error {
 		return err
 	}
 
-	// Only attempt to steal one shard at at time, to allow for linear convergence
+	// Only attempt to steal one shard at time, to allow for linear convergence
 	if w.shardStealInProgress {
 		shardInfo := make(map[string]bool)
 		err := w.getShardIDs("", shardInfo)
@@ -418,12 +417,12 @@ func (w *Worker) rebalance() error {
 		log.Debugf("We have enough shards, not attempting to steal any. workerID: %s", w.workerID)
 		return nil
 	}
-	maxShards := int(optimalShards)
+
 	var workerSteal string
 	for worker, shards := range workers {
-		if worker != w.workerID && len(shards) > maxShards {
+		if worker != w.workerID && len(shards) > optimalShards {
 			workerSteal = worker
-			maxShards = len(shards)
+			optimalShards = len(shards)
 		}
 	}
 	// Not all shards are allocated so fallback to default shard allocation mechanisms
@@ -434,7 +433,8 @@ func (w *Worker) rebalance() error {
 
 	// Steal a random shard from the worker with the most shards
 	w.shardStealInProgress = true
-	randIndex := rand.Intn(len(workers[workerSteal]))
+	rnd, _ := rand.Int(rand.Reader, big.NewInt(int64(len(workers[workerSteal]))))
+	randIndex := int(rnd.Int64())
 	shardToSteal := workers[workerSteal][randIndex]
 	log.Debugf("Stealing shard %s from %s", shardToSteal, workerSteal)
 

go.mod

@@ -1,22 +1,34 @@
 module github.com/vmware/vmware-go-kcl
 
+go 1.17
+
 require (
-	github.com/BurntSushi/toml v0.3.1 // indirect
-	github.com/aws/aws-sdk-go v1.34.8
-	github.com/awslabs/kinesis-aggregation/go v0.0.0-20201211133042-142dfe1d7a6d
-	github.com/golang/protobuf v1.3.1
-	github.com/google/uuid v1.1.1
-	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
-	github.com/prometheus/client_golang v0.9.3
-	github.com/prometheus/common v0.4.1
-	github.com/prometheus/procfs v0.0.0-20190523193104-a7aeb8df3389 // indirect
+	github.com/aws/aws-sdk-go v1.41.7
+	github.com/awslabs/kinesis-aggregation/go v0.0.0-20210630091500-54e17340d32f
+	github.com/golang/protobuf v1.5.2
+	github.com/google/uuid v1.3.0
+	github.com/prometheus/client_golang v1.11.0
+	github.com/prometheus/common v0.32.1
 	github.com/rs/zerolog v1.25.0
-	github.com/sirupsen/logrus v1.4.2
-	github.com/stretchr/testify v1.5.1
-	go.uber.org/atomic v1.4.0 // indirect
-	go.uber.org/multierr v1.2.0 // indirect
-	go.uber.org/zap v1.11.0
+	github.com/sirupsen/logrus v1.8.1
+	github.com/stretchr/testify v1.7.0
+	go.uber.org/zap v1.19.1
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 )
 
-go 1.13
+require (
+	github.com/BurntSushi/toml v0.4.1 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/procfs v0.7.3 // indirect
+	go.uber.org/atomic v1.9.0 // indirect
+	go.uber.org/multierr v1.7.0 // indirect
+	golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac // indirect
+	google.golang.org/protobuf v1.27.1 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+)