feat: add WatchChan method

This method creates a watcher for a key or prefix and unmarshals events into channel.

Author: zbindenren

.github/workflows/codeql-analysis.yml

@@ -14,6 +14,10 @@ on:
   schedule:
     - cron: '0 5 * * 4'
 
+concurrency:
+  group: '${{ github.head_ref || github.ref }}'
+  cancel-in-progress: true
+
 jobs:
   analyze:
     name: Analyze

.github/workflows/push.yml

@@ -1,6 +1,10 @@
 name: build
 on: [push, pull_request]
 
+concurrency:
+  group: '${{ github.head_ref || github.ref }}'
+  cancel-in-progress: true
+
 jobs:
   lint:
     runs-on: ubuntu-latest

.golangci.yml

@@ -1,19 +1,20 @@
-# options for analysis running
+---
 run:
-  tests: true
-  timeout: 3m
-
-
-# all available settings of specific linters
+  tests: false
+  skip-dirs:
+    - .github
+    - build
+    - web
+    - .go
 linters-settings:
   dupl:
-    threshold: 150
+    threshold: 100
   funlen:
     lines: 100
     statements: 50
   goconst:
     min-len: 2
-    min-occurrences: 3
+    min-occurrences: 2
   gocritic:
     enabled-tags:
       - diagnostic
@@ -22,22 +23,26 @@ linters-settings:
       - performance
       - style
     disabled-checks:
-      - whyNoLint
+      - hugeParam
   gocyclo:
-    min-complexity: 15
-  goimports:
-    local-prefixes: github.com/ewohltman/ephemeral-roles
+    min-complexity: 20
   revive:
     min-confidence: 0.8
   govet:
     check-shadowing: true
+  lll:
+    line-length: 140
   misspell:
     locale: US
+  nolintlint:
+    allow-leading-space: false
+    allow-no-explanation:
+      - gocognit
+      - funlen
+      - gocyclo
 
 linters:
-  disable:
-    - gomnd
-    - lll
+  disable-all: true
   enable:
     - bodyclose
     - deadcode
@@ -46,13 +51,13 @@ linters:
     - dupl
     - errcheck
     - funlen
+    - nolintlint
     - gochecknoglobals
     - gochecknoinits
     - gocognit
     - goconst
     - gocritic
     - gocyclo
-    - godox
     - gofmt
     - goimports
     - revive
@@ -77,11 +82,11 @@ linters:
     - whitespace
     - wsl
 
-
 issues:
   exclude:
     # Very commonly not checked.
-    - 'Error return value of .(l.Sync|.*Close|.*Flush|os\.Remove(All)?|os\.(Un)?Setenv). is not checked'
+    - 'Error return value of .(l.Sync|.*Close|.*.Write|.*Flush|os\.Remove(All)?|os\.(Un)?Setenv). is not checked'
+    - 'G104:.*'
     - 'exported method (.*\.MarshalJSON|.*\.UnmarshalJSON) should have comment or be unexported'
     - 'shadow: declaration of "err" shadows declaration.*'
   max-same-issues: 0

encoding.go

@@ -8,15 +8,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestUnmarshalList(t *testing.T) {
-	count := 1000
-	b := newMockBackend(count)
-	l := []testStruct{}
-	err := UnmarshalJSONList(&l, "", b)
-	require.NoError(t, err)
-	assert.Len(t, l, count)
-}
-
 func TestUnmarshal(t *testing.T) {
 	count := 1000
 	b := newMockBackend(count)
@@ -34,19 +25,6 @@ func TestUnmarshal(t *testing.T) {
 	assert.Len(t, l, count)
 }
 
-func benchmarkUnmarshalList(i int, b *testing.B) {
-	b.StopTimer()
-
-	back := newMockBackend(i)
-
-	b.ReportAllocs()
-	b.StartTimer()
-
-	for n := 0; n < b.N; n++ {
-		l := []testStruct{}
-		_ = UnmarshalJSONList(&l, "", back)
-	}
-}
 func benchmarkUnmarshal(i int, b *testing.B) {
 	b.StopTimer()
 
@@ -72,26 +50,16 @@ func benchmarkUnmarshal(i int, b *testing.B) {
 
 /*
 BenchmarkUnmarshal10-4                    200000            194390 ns/op
-BenchmarkUnmarshalList10-4                300000            201912 ns/op
 BenchmarkUnmarshal100-4                    20000           2076871 ns/op
-BenchmarkUnmarshalList100-4                20000           2110142 ns/op
 BenchmarkUnmarshal1000-4                    2000          21495289 ns/op
-BenchmarkUnmarshalList1000-4                2000          21822339 ns/op
 BenchmarkUnmarshal10000-4                    200         233705937 ns/op
-BenchmarkUnmarshalList10000-4                200         211692563 ns/op
 BenchmarkUnmarshal100000-4                    20        2317840686 ns/op
-BenchmarkUnmarshalList100000-4                20        2118017559 ns/op
 */
-func BenchmarkUnmarshal10(b *testing.B)         { benchmarkUnmarshal(10, b) }
-func BenchmarkUnmarshalList10(b *testing.B)     { benchmarkUnmarshalList(10, b) }
-func BenchmarkUnmarshal100(b *testing.B)        { benchmarkUnmarshal(100, b) }
-func BenchmarkUnmarshalList100(b *testing.B)    { benchmarkUnmarshalList(100, b) }
-func BenchmarkUnmarshal1000(b *testing.B)       { benchmarkUnmarshal(1000, b) }
-func BenchmarkUnmarshalList1000(b *testing.B)   { benchmarkUnmarshalList(1000, b) }
-func BenchmarkUnmarshal10000(b *testing.B)      { benchmarkUnmarshal(10000, b) }
-func BenchmarkUnmarshalList10000(b *testing.B)  { benchmarkUnmarshalList(10000, b) }
-func BenchmarkUnmarshal100000(b *testing.B)     { benchmarkUnmarshal(100000, b) }
-func BenchmarkUnmarshalList100000(b *testing.B) { benchmarkUnmarshalList(100000, b) }
+func BenchmarkUnmarshal10(b *testing.B)     { benchmarkUnmarshal(10, b) }
+func BenchmarkUnmarshal100(b *testing.B)    { benchmarkUnmarshal(100, b) }
+func BenchmarkUnmarshal1000(b *testing.B)   { benchmarkUnmarshal(1000, b) }
+func BenchmarkUnmarshal10000(b *testing.B)  { benchmarkUnmarshal(10000, b) }
+func BenchmarkUnmarshal100000(b *testing.B) { benchmarkUnmarshal(100000, b) }
 
 type mockBackend struct {
 	entries []Entry
@@ -136,6 +104,10 @@ func (mockBackend) Close() error {
 	return nil
 }
 
+func (m *mockBackend) WatchChan(string, interface{}, chan error, ...WatchOption) (WatchStarter, error) {
+	return nil, nil
+}
+
 func newMockBackend(max int) Backend {
 	entries := make([]Entry, 0, max)
 	for i := 0; i < max; i++ {
@@ -188,5 +160,4 @@ type testStruct struct {
 	} `json:"commands"`
 }
 
-//nolint:gochecknoglobals
-var data = `{"id":"admin:lslb-pool","organization":"admin","name":"lslb-pool","description":"","user":"","group":"","shell":"bash","timeout":0,"interval":300000000000,"constraints":{"os":["linux"],"host":["p1-linux-mlsu005","p1-linux-mlsu006"],"files":["/usr/bin/lslb"]},"discovery":{"script":{"data":"echo \"fake-pool\"\n# sudo /usr/bin/lslb pools","rediscover":false}},"tags":null,"commands":{"start":{"shell":"/bin/true","timeout":"10s","rediscover":false},"status":{"shell":"/bin/true","timeout":"10s","rediscover":false},"stop":{"shell":"/bin/true","timeout":"10s","rediscover":false}}}`
+const data = `{"id":"admin:lslb-pool","organization":"admin","name":"lslb-pool","description":"","user":"","group":"","shell":"bash","timeout":0,"interval":300000000000,"constraints":{"os":["linux"],"host":["p1-linux-mlsu005","p1-linux-mlsu006"],"files":["/usr/bin/lslb"]},"discovery":{"script":{"data":"echo \"fake-pool\"\n# sudo /usr/bin/lslb pools","rediscover":false}},"tags":null,"commands":{"start":{"shell":"/bin/true","timeout":"10s","rediscover":false},"status":{"shell":"/bin/true","timeout":"10s","rediscover":false},"stop":{"shell":"/bin/true","timeout":"10s","rediscover":false}}}`

encoding_test.go

@@ -17,8 +17,9 @@ import (
 
 // Constants
 const (
-	DfltSeparator    = '/'
-	DfltKeepAliveTTL = 30 * time.Second
+	DfltSeparator          = '/'
+	DfltKeepAliveTTL       = 30 * time.Second
+	DfltWatchNotifyTimeout = 5 * time.Second
 )
 
 // verify that *Backend implements store.Backend
@@ -28,8 +29,9 @@ var _ store.Backend = &Backend{}
 // has to be used.
 func New(opts ...Opt) (*Backend, error) {
 	e := Backend{
-		separator:  DfltSeparator,
-		errHandler: func(err error) error { return err },
+		separator:          DfltSeparator,
+		errHandler:         func(err error) error { return err },
+		watchNotifyTimeout: DfltWatchNotifyTimeout,
 	}
 
 	for _, option := range opts {
@@ -121,6 +123,20 @@ func WithRequestTimeout(timeout time.Duration) Opt {
 	}
 }
 
+// WithWatchNotifyTimeout determines how long to wait for a
+// watch notify create event. Default is 5s.
+func WithWatchNotifyTimeout(timeout time.Duration) Opt {
+	return func(e *Backend) error {
+		if timeout <= 0 {
+			return errors.New("watch notify timeout cannot be <= 0")
+		}
+
+		e.watchNotifyTimeout = timeout
+
+		return nil
+	}
+}
+
 // WithUsername is an option to configure etcd username.
 func WithUsername(username string) Opt {
 	return func(e *Backend) error {

etcd/config.go

@@ -33,6 +33,7 @@ type Backend struct {
 	ctx                  context.Context
 	errHandler           store.ErrorFunc
 	dialOptions          []grpc.DialOption
+	watchNotifyTimeout   time.Duration
 }
 
 // AbsKey will convert a relativ key in a absolute key
@@ -82,7 +83,6 @@ func (e *Backend) KeyLeaf(key string) string {
 }
 
 // Put is used to insert or update an entry
-//nolint:gocyclo
 func (e *Backend) Put(entry *store.Entry, ops ...store.PutOption) (bool, error) {
 	opts := &store.PutOptions{}
 
@@ -145,7 +145,6 @@ func (e *Backend) Put(entry *store.Entry, ops ...store.PutOption) (bool, error)
 }
 
 // Get is used to fetch an one ore many entries.
-//nolint:gocyclo
 func (e *Backend) Get(key string, ops ...store.GetOption) ([]store.Entry, error) {
 	opts := &store.GetOptions{}
 

etcd/etcdv3.go

@@ -465,40 +465,6 @@ func TestMarshal(t *testing.T) {
 	}
 }
 
-func TestMarshal2(t *testing.T) {
-	integration.BeforeTestExternal(t)
-
-	for _, p := range []string{"", "root"} {
-		opts := []Opt{}
-		if p != "" {
-			opts = append(opts, WithPrefix(p))
-		}
-
-		b, _, teardown := setupTestStore(t, false, opts)
-
-		type testStruct struct {
-			Value1 string
-			Value2 int
-		}
-
-		exp := testStruct{
-			Value1: "value1",
-			Value2: 42,
-		}
-		key := "key"
-		value, err := json.Marshal(exp)
-		require.NoError(t, err)
-		_, err = b.Put(&store.Entry{Key: key, Value: value})
-		require.NoError(t, err)
-
-		bla := []testStruct{}
-		err = store.UnmarshalJSONList(&bla, key, b)
-		require.NoError(t, err)
-
-		teardown()
-	}
-}
-
 /*
 SetupTestStore starts an embedded etcd server.
 The returned function shall be used as teardown function.