Fixed ClusterConfig loading for boolean values (#555)

streamnative · Oct 21, 2024 · 3402e22 · 3402e22
1 parent 7c929b5
commit 3402e22
Show file tree

Hide file tree

Showing 3 changed files with 55 additions and 22 deletions.
diff --git a/cmd/coordinator/cmd.go b/cmd/coordinator/cmd.go
@@ -15,11 +15,13 @@
 package coordinator
 
 import (
-	"errors"
 	"io"
 	"log/slog"
 	"strings"
 
+	"github.com/mitchellh/mapstructure"
+	"github.com/pkg/errors"
+
 	"github.com/fsnotify/fsnotify"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -128,8 +130,12 @@ func loadClusterConfig(v *viper.Viper) (model.ClusterConfig, error) {
 		return cc, err
 	}
 
-	if err := v.Unmarshal(&cc); err != nil {
-		return cc, err
+	if err := v.Unmarshal(&cc, viper.DecodeHook(mapstructure.ComposeDecodeHookFunc(
+		common.OptBooleanViperHook(),
+		mapstructure.StringToTimeDurationHookFunc(), // default hook
+		mapstructure.StringToSliceHookFunc(","),     // default hook
+	))); err != nil {
+		return cc, errors.Wrap(err, "failed to load cluster config")
 	}
 
 	return cc, nil

diff --git a/cmd/coordinator/cmd_test.go b/cmd/coordinator/cmd_test.go
@@ -33,9 +33,10 @@ import (
 func TestCmd(t *testing.T) {
 	clusterConfig := model.ClusterConfig{
 		Namespaces: []model.NamespaceConfig{{
-			Name:              common.DefaultNamespace,
-			ReplicationFactor: 1,
-			InitialShardCount: 2,
+			Name:                 common.DefaultNamespace,
+			ReplicationFactor:    1,
+			InitialShardCount:    2,
+			NotificationsEnabled: common.Bool(false),
 		}},
 		Servers: []model.ServerAddress{{
 			Public:   "public:1234",
@@ -47,7 +48,6 @@ func TestCmd(t *testing.T) {
 	assert.NoError(t, err)
 
 	name := "config.yaml"
-
 	err = os.WriteFile(name, bytes, os.ModePerm)
 	assert.NoError(t, err)
 
@@ -68,9 +68,10 @@ func TestCmd(t *testing.T) {
 			MetadataProviderImpl: coordinator.File,
 		}, model.ClusterConfig{
 			Namespaces: []model.NamespaceConfig{{
-				Name:              common.DefaultNamespace,
-				ReplicationFactor: 1,
-				InitialShardCount: 2,
+				Name:                 common.DefaultNamespace,
+				ReplicationFactor:    1,
+				InitialShardCount:    2,
+				NotificationsEnabled: common.Bool(false),
 			}},
 			Servers: []model.ServerAddress{{
 				Public:   "public:1234",
@@ -84,9 +85,10 @@ func TestCmd(t *testing.T) {
 			MetadataProviderImpl: coordinator.File,
 		}, model.ClusterConfig{
 			Namespaces: []model.NamespaceConfig{{
-				Name:              common.DefaultNamespace,
-				ReplicationFactor: 1,
-				InitialShardCount: 2,
+				Name:                 common.DefaultNamespace,
+				ReplicationFactor:    1,
+				InitialShardCount:    2,
+				NotificationsEnabled: common.Bool(false),
 			}},
 			Servers: []model.ServerAddress{{
 				Public:   "public:1234",
@@ -100,9 +102,10 @@ func TestCmd(t *testing.T) {
 			MetadataProviderImpl: coordinator.File,
 		}, model.ClusterConfig{
 			Namespaces: []model.NamespaceConfig{{
-				Name:              common.DefaultNamespace,
-				ReplicationFactor: 1,
-				InitialShardCount: 2,
+				Name:                 common.DefaultNamespace,
+				ReplicationFactor:    1,
+				InitialShardCount:    2,
+				NotificationsEnabled: common.Bool(false),
 			}},
 			Servers: []model.ServerAddress{{
 				Public:   "public:1234",
@@ -116,9 +119,10 @@ func TestCmd(t *testing.T) {
 			MetadataProviderImpl: coordinator.File,
 		}, model.ClusterConfig{
 			Namespaces: []model.NamespaceConfig{{
-				Name:              common.DefaultNamespace,
-				ReplicationFactor: 1,
-				InitialShardCount: 2,
+				Name:                 common.DefaultNamespace,
+				ReplicationFactor:    1,
+				InitialShardCount:    2,
+				NotificationsEnabled: common.Bool(false),
 			}},
 			Servers: []model.ServerAddress{{
 				Public:   "public:1234",
@@ -132,9 +136,10 @@ func TestCmd(t *testing.T) {
 			MetadataProviderImpl: coordinator.File,
 		}, model.ClusterConfig{
 			Namespaces: []model.NamespaceConfig{{
-				Name:              common.DefaultNamespace,
-				ReplicationFactor: 1,
-				InitialShardCount: 2,
+				Name:                 common.DefaultNamespace,
+				ReplicationFactor:    1,
+				InitialShardCount:    2,
+				NotificationsEnabled: common.Bool(false),
 			}},
 			Servers: []model.ServerAddress{{
 				Public:   "public:1234",

diff --git a/common/opt_json.go b/common/opt_json.go
@@ -16,6 +16,9 @@ package common
 
 import (
 	"encoding/json"
+	"reflect"
+
+	"github.com/mitchellh/mapstructure"
 
 	"github.com/pkg/errors"
 )
@@ -40,6 +43,10 @@ func (o *OptBooleanDefaultTrue) MarshalJSON() ([]byte, error) {
 	return json.Marshal(o.val)
 }
 
+func (o OptBooleanDefaultTrue) MarshalYAML() (any, error) {
+	return o.val, nil
+}
+
 var trueVal = true
 var falseVal = false
 
@@ -62,3 +69,18 @@ func (o *OptBooleanDefaultTrue) UnmarshalJSON(data []byte) error {
 
 	return errors.New("invalid boolean value: " + s)
 }
+
+func OptBooleanViperHook() mapstructure.DecodeHookFuncType {
+	return func(_ reflect.Type, t reflect.Type, data any) (any, error) {
+		if t == reflect.TypeOf(OptBooleanDefaultTrue{}) {
+			b, ok := data.(bool)
+			if !ok {
+				return nil, errors.Errorf("invalid boolean value: '%v'", data)
+			}
+
+			return Bool(b), nil
+		}
+
+		return data, nil
+	}
+}