manager, config: watch and apply the latest log configuration online

cmd/tiproxy/main.go

@@ -25,7 +25,6 @@ import (
 	"github.com/pingcap/TiProxy/lib/util/waitgroup"
 	"github.com/pingcap/TiProxy/pkg/server"
 	"github.com/spf13/cobra"
-	"go.uber.org/zap"
 )
 
 func main() {
@@ -57,19 +56,7 @@ func main() {
 			cfg.Log.Level = *logLevel
 		}
 
-		zapcfg := zap.NewDevelopmentConfig()
-		zapcfg.Encoding = cfg.Log.Encoder
-		zapcfg.DisableStacktrace = true
-		if level, err := zap.ParseAtomicLevel(cfg.Log.Level); err == nil {
-			zapcfg.Level = level
-		}
-		logger, err := zapcfg.Build()
-		if err != nil {
-			return err
-		}
-		logger = logger.Named("main")
-
-		srv, err := server.NewServer(cmd.Context(), cfg, logger, *pubAddr)
+		srv, err := server.NewServer(cmd.Context(), cfg, *pubAddr)
 		if err != nil {
 			return errors.Wrapf(err, "fail to create server")
 		}
@@ -80,8 +67,6 @@ func main() {
 		<-cmd.Context().Done()
 		if e := srv.Close(); e != nil {
 			err = errors.Wrapf(err, "shutdown with errors")
-		} else {
-			logger.Info("gracefully shutdown")
 		}
 
 		wg.Wait()

lib/cli/config.go

@@ -27,10 +27,11 @@ const (
 	configPrefix = "/api/admin/config"
 )
 
-func GetConfigProxyCmd(ctx *Context) *cobra.Command {
+func getConfigCmd(ctx *Context, pathSuffix string) *cobra.Command {
 	rootCmd := &cobra.Command{
-		Use: "proxy",
+		Use: pathSuffix,
 	}
+	path := fmt.Sprintf("%s/%s", configPrefix, pathSuffix)
 
 	// set config proxy
 	{
@@ -50,7 +51,7 @@ func GetConfigProxyCmd(ctx *Context) *cobra.Command {
 				b = os.Stdin
 			}
 
-			resp, err := doRequest(cmd.Context(), ctx, http.MethodPut, fmt.Sprintf("%s/proxy", configPrefix), b)
+			resp, err := doRequest(cmd.Context(), ctx, http.MethodPut, path, b)
 			if err != nil {
 				return err
 			}
@@ -67,7 +68,7 @@ func GetConfigProxyCmd(ctx *Context) *cobra.Command {
 			Use: "get",
 		}
 		getProxy.RunE = func(cmd *cobra.Command, args []string) error {
-			resp, err := doRequest(cmd.Context(), ctx, http.MethodGet, fmt.Sprintf("%s/proxy", configPrefix), nil)
+			resp, err := doRequest(cmd.Context(), ctx, http.MethodGet, path, nil)
 			if err != nil {
 				return err
 			}
@@ -86,6 +87,7 @@ func GetConfigCmd(ctx *Context) *cobra.Command {
 		Use:   "config",
 		Short: "",
 	}
-	rootCmd.AddCommand(GetConfigProxyCmd(ctx))
+	rootCmd.AddCommand(getConfigCmd(ctx, "proxy"))
+	rootCmd.AddCommand(getConfigCmd(ctx, "log"))
 	return rootCmd
 }

lib/config/proxy.go

@@ -62,12 +62,16 @@ type Advance struct {
 	IgnoreWrongNamespace bool   `yaml:"ignore-wrong-namespace,omitempty" toml:"ignore-wrong-namespace,omitempty" json:"ignore-wrong-namespace,omitempty"`
 }
 
-type Log struct {
+type LogOnline struct {
 	Level   string  `yaml:"level,omitempty" toml:"level,omitempty" json:"level,omitempty"`
-	Encoder string  `yaml:"encoder,omitempty" toml:"encoder,omitempty" json:"encoder,omitempty"`
 	LogFile LogFile `yaml:"log-file,omitempty" toml:"log-file,omitempty" json:"log-file,omitempty"`
 }
 
+type Log struct {
+	Encoder   string `yaml:"encoder,omitempty" toml:"encoder,omitempty" json:"encoder,omitempty"`
+	LogOnline `yaml:",inline" toml:",inline" json:",inline"`
+}
+
 type LogFile struct {
 	Filename   string `yaml:"filename,omitempty" toml:"filename,omitempty" json:"filename,omitempty"`
 	MaxSize    int    `yaml:"max-size,omitempty" toml:"max-size,omitempty" json:"max-size,omitempty"`

lib/config/proxy_test.go

@@ -46,13 +46,15 @@ var testProxyConfig = Config{
 		MetricsInterval: 15,
 	},
 	Log: Log{
-		Level:   "info",
 		Encoder: "tidb",
-		LogFile: LogFile{
-			Filename:   ".",
-			MaxSize:    10,
-			MaxDays:    1,
-			MaxBackups: 1,
+		LogOnline: LogOnline{
+			Level: "info",
+			LogFile: LogFile{
+				Filename:   ".",
+				MaxSize:    10,
+				MaxDays:    1,
+				MaxBackups: 1,
+			},
 		},
 	},
 	Security: Security{

pkg/manager/config/config.go

@@ -0,0 +1,186 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"context"
+	"encoding/json"
+	"reflect"
+
+	"github.com/pingcap/TiProxy/lib/config"
+	"github.com/pingcap/TiProxy/lib/util/errors"
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	"go.uber.org/zap"
+)
+
+type mKeyType reflect.Type
+
+type OnlineCfgTypes interface {
+	config.ProxyServerOnline | config.LogOnline
+}
+
+type imeta interface {
+	getPrefix() string
+	unmarshal(bytes []byte) (any, error)
+	addToCh(any)
+	getInitial(cfg *config.Config) any
+}
+
+type meta[T OnlineCfgTypes] struct {
+	prefix   string
+	initFunc func(cfg *config.Config) T
+	ch       chan *T
+}
+
+func newMeta[T OnlineCfgTypes](prefix string, initFunc func(cfg *config.Config) T) *meta[T] {
+	return &meta[T]{
+		prefix:   prefix,
+		initFunc: initFunc,
+		ch:       make(chan *T, 1),
+	}
+}
+
+func (m meta[T]) unmarshal(bytes []byte) (any, error) {
+	var t T
+	err := json.Unmarshal(bytes, &t)
+	return &t, err
+}
+
+func (m meta[T]) getPrefix() string {
+	return m.prefix
+}
+
+func (m meta[T]) addToCh(obj any) {
+	m.ch <- obj.(*T)
+}
+
+func (m meta[T]) getInitial(cfg *config.Config) any {
+	return m.initFunc(cfg)
+}
+
+func getMetaKey[T OnlineCfgTypes]() mKeyType {
+	return reflect.TypeOf(new(T))
+}
+
+func getMetaKeyByObj[T OnlineCfgTypes](t *T) mKeyType {
+	return reflect.TypeOf(t)
+}
+
+func (e *ConfigManager) initMetas() {
+	e.metas = map[mKeyType]imeta{
+		getMetaKey[config.ProxyServerOnline](): newMeta(pathPrefixProxyServer, func(cfg *config.Config) config.ProxyServerOnline {
+			return cfg.Proxy.ProxyServerOnline
+		}),
+		getMetaKey[config.LogOnline](): newMeta(pathPrefixLog, func(cfg *config.Config) config.LogOnline {
+			return cfg.Log.LogOnline
+		}),
+	}
+}
+
+func (e *ConfigManager) watchConfig(ctx context.Context, cfg *config.Config) error {
+	for _, m := range e.metas {
+		if err := func(m imeta) error {
+			_, err := e.get(ctx, m.getPrefix(), "")
+			if err != nil && errors.Is(err, ErrNoOrMultiResults) {
+				value, err := json.Marshal(m.getInitial(cfg))
+				if err != nil {
+					return err
+				}
+				if err = e.set(ctx, m.getPrefix(), "", string(value)); err != nil {
+					return err
+				}
+			}
+			e.watch(ctx, m.getPrefix(), "", func(logger *zap.Logger, evt mvccpb.Event) {
+				if obj, err := m.unmarshal(evt.Kv.Value); err != nil {
+					logger.Warn("failed unmarshal proxy config", zap.Error(err))
+					return
+				} else {
+					m.addToCh(obj)
+				}
+			})
+			return nil
+		}(m); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (e *ConfigManager) getCfg(ctx context.Context, metaKey reflect.Type) (any, error) {
+	m := e.metas[metaKey]
+	val, err := e.get(ctx, m.getPrefix(), "")
+	if err != nil {
+		return nil, err
+	}
+	return m.unmarshal(val.Value)
+}
+
+func (e *ConfigManager) setCfg(ctx context.Context, metaKey mKeyType, obj any) error {
+	m := e.metas[metaKey]
+	value, err := json.Marshal(obj)
+	if err != nil {
+		return err
+	}
+	return e.set(ctx, m.getPrefix(), "", string(value))
+}
+
+// GetConfig queries the configuration from the config center.
+func GetConfig[T OnlineCfgTypes](ctx context.Context, e *ConfigManager, t *T) error {
+	obj, err := e.getCfg(ctx, getMetaKeyByObj(t))
+	if err != nil {
+		return err
+	}
+	*t = *obj.(*T)
+	return nil
+}
+
+// SetConfig sets a configuration to the config center.
+func SetConfig[T OnlineCfgTypes](ctx context.Context, e *ConfigManager, t *T) error {
+	return e.setCfg(ctx, getMetaKeyByObj(t), t)
+}
+
+// GetCfgWatch returns the channel that contains updated configuration.
+func GetCfgWatch[T OnlineCfgTypes](e *ConfigManager) chan *T {
+	mt := e.metas[getMetaKey[T]()].(*meta[T])
+	return mt.ch
+}
+
+func (e *ConfigManager) GetProxyConfigWatch() <-chan *config.ProxyServerOnline {
+	return GetCfgWatch[config.ProxyServerOnline](e)
+}
+
+func (e *ConfigManager) GetProxyConfig(ctx context.Context) (*config.ProxyServerOnline, error) {
+	var pso config.ProxyServerOnline
+	err := GetConfig(ctx, e, &pso)
+	return &pso, err
+}
+
+func (e *ConfigManager) SetProxyConfig(ctx context.Context, proxy *config.ProxyServerOnline) error {
+	return SetConfig(ctx, e, proxy)
+}
+
+func (e *ConfigManager) GetLogConfigWatch() <-chan *config.LogOnline {
+	return GetCfgWatch[config.LogOnline](e)
+}
+
+func (e *ConfigManager) GetLogConfig(ctx context.Context) (*config.LogOnline, error) {
+	var co config.LogOnline
+	err := GetConfig(ctx, e, &co)
+	return &co, err
+}
+
+func (e *ConfigManager) SetLogConfig(ctx context.Context, log *config.LogOnline) error {
+	return SetConfig(ctx, e, log)
+}

pkg/manager/config/proxy_test.go → pkg/manager/config/config_test.go

@@ -15,6 +15,7 @@
 package config
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -23,8 +24,6 @@ import (
 )
 
 func TestProxyConfig(t *testing.T) {
-	cfgmgr, ctx := testConfigManager(t, &config.Config{})
-
 	cases := []*config.ProxyServerOnline{
 		{
 			MaxConnections: 1,
@@ -44,11 +43,54 @@ func TestProxyConfig(t *testing.T) {
 		},
 	}
 
-	ch := cfgmgr.GetProxyConfigWatch()
-	require.Equal(t, <-ch, &config.ProxyServerOnline{})
+	testWatch(t, cases, func(cfgmgr *ConfigManager) <-chan *config.ProxyServerOnline {
+		return cfgmgr.GetProxyConfigWatch()
+	}, func(ctx context.Context, cfgmgr *ConfigManager, tc *config.ProxyServerOnline) error {
+		return cfgmgr.SetProxyConfig(ctx, tc)
+	})
+}
+
+func TestLogConfig(t *testing.T) {
+	cases := []*config.LogOnline{
+		{
+			Level: "info",
+			LogFile: config.LogFile{
+				Filename:   "proxy.log",
+				MaxSize:    100,
+				MaxDays:    10,
+				MaxBackups: 10,
+			},
+		},
+		{
+			Level: "debug",
+			LogFile: config.LogFile{
+				Filename:   "l.log",
+				MaxSize:    1,
+				MaxDays:    1,
+				MaxBackups: 1,
+			},
+		},
+		{
+			Level:   "",
+			LogFile: config.LogFile{},
+		},
+	}
+
+	testWatch(t, cases, func(cfgmgr *ConfigManager) <-chan *config.LogOnline {
+		return cfgmgr.GetLogConfigWatch()
+	}, func(ctx context.Context, cfgmgr *ConfigManager, tc *config.LogOnline) error {
+		return cfgmgr.SetLogConfig(ctx, tc)
+	})
+}
+
+func testWatch[T OnlineCfgTypes](t *testing.T, cases []*T, getWatch func(*ConfigManager) <-chan *T,
+	setConfig func(context.Context, *ConfigManager, *T) error) {
+	cfgmgr, ctx := testConfigManager(t, &config.Config{})
+	ch := getWatch(cfgmgr)
+	require.Equal(t, <-ch, new(T))
 
 	for _, tc := range cases {
-		require.NoError(t, cfgmgr.SetProxyConfig(ctx, tc))
+		require.NoError(t, setConfig(ctx, cfgmgr, tc))
 		select {
 		case <-time.After(5 * time.Second):
 			t.Fatalf("\n\ntimeout waiting chan\n\n")