feat: 初始化配置中心项目
This commit is contained in:
@@ -0,0 +1,291 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Client is the config center client
|
||||
type Client struct {
|
||||
config *Config
|
||||
cache *Cache
|
||||
listener *Listener
|
||||
configs map[string]string
|
||||
secret string
|
||||
globalSecret string
|
||||
lastTime int64
|
||||
mu sync.RWMutex
|
||||
httpClient *http.Client
|
||||
stopCh chan struct{}
|
||||
stopOnce sync.Once
|
||||
started bool
|
||||
}
|
||||
|
||||
// New creates a new config client
|
||||
func New(config *Config) *Client {
|
||||
if config.Version == "" {
|
||||
config.Version = "latest"
|
||||
}
|
||||
if config.SyncInterval <= 0 {
|
||||
config.SyncInterval = 60
|
||||
}
|
||||
if config.CacheDir == "" {
|
||||
config.CacheDir = "."
|
||||
}
|
||||
|
||||
return &Client{
|
||||
config: config,
|
||||
cache: NewCache(config.CacheDir),
|
||||
listener: NewListener(),
|
||||
configs: make(map[string]string),
|
||||
httpClient: &http.Client{Timeout: 10 * time.Second},
|
||||
stopCh: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// OnChange registers a config change handler
|
||||
func (c *Client) OnChange(handler ChangeHandler) {
|
||||
c.listener.OnChange(handler)
|
||||
}
|
||||
|
||||
// Get returns a config value by key
|
||||
func (c *Client) Get(key string) string {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
return c.configs[key]
|
||||
}
|
||||
|
||||
// GetAll returns all configs
|
||||
func (c *Client) GetAll() map[string]string {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
result := make(map[string]string, len(c.configs))
|
||||
for k, v := range c.configs {
|
||||
result[k] = v
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// Start initializes the client and starts sync loop
|
||||
func (c *Client) Start() error {
|
||||
c.mu.Lock()
|
||||
if c.started {
|
||||
c.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
c.started = true
|
||||
c.mu.Unlock()
|
||||
|
||||
// fetch all configs
|
||||
if err := c.fetchAll(); err != nil {
|
||||
log.Printf("[ConfigClient] fetch all configs error: %v, trying cache", err)
|
||||
if cached, cacheErr := c.cache.Read(); cacheErr == nil {
|
||||
c.applyConfigs(cached)
|
||||
log.Printf("[ConfigClient] loaded %d configs from cache", len(c.configs))
|
||||
} else {
|
||||
return fmt.Errorf("fetch configs failed and no cache: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// start sync loop
|
||||
go c.syncLoop()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop stops the sync loop
|
||||
func (c *Client) Stop() {
|
||||
c.stopOnce.Do(func() {
|
||||
close(c.stopCh)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *Client) fetchAll() error {
|
||||
url := fmt.Sprintf("%s/api/fetchAllConfigs?appName=%s&env=%s&version=%s&ignoreGlobal=%v",
|
||||
c.config.BaseURL, c.config.AppName, c.config.Env, c.config.Version, c.config.IgnoreGlobal)
|
||||
|
||||
req, err := http.NewRequest("GET", url, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("server returned HTTP %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var result map[string]interface{}
|
||||
if err := json.Unmarshal(body, &result); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// check error response
|
||||
if code, ok := result["code"]; ok {
|
||||
codeInt, _ := code.(float64)
|
||||
if codeInt != 0 {
|
||||
msg, _ := result["msg"].(string)
|
||||
return fmt.Errorf("server error: %s", msg)
|
||||
}
|
||||
}
|
||||
|
||||
// extract secrets
|
||||
if s, ok := result["kenaito.config.encrypt-secret"]; ok {
|
||||
c.secret, _ = s.(string)
|
||||
}
|
||||
if s, ok := result["kenaito.config.global-encrypt-secret"]; ok {
|
||||
c.globalSecret, _ = s.(string)
|
||||
}
|
||||
|
||||
// extract timestamp
|
||||
if ts, ok := result["latest.fetch.timestamp"]; ok {
|
||||
if tsStr, ok := ts.(string); ok {
|
||||
fmt.Sscanf(tsStr, "%d", &c.lastTime)
|
||||
}
|
||||
}
|
||||
|
||||
// save to cache
|
||||
c.cache.Write(result)
|
||||
|
||||
// apply configs
|
||||
c.applyConfigs(result)
|
||||
|
||||
log.Printf("[ConfigClient] fetched %d configs", len(c.configs))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Client) fetchChanged() error {
|
||||
url := fmt.Sprintf("%s/api/fetchChangedConfigs", c.config.BaseURL)
|
||||
|
||||
payload := map[string]interface{}{
|
||||
"appName": c.config.AppName,
|
||||
"env": c.config.Env,
|
||||
"version": c.config.Version,
|
||||
"lastTime": c.lastTime,
|
||||
"ignoreGlobal": c.config.IgnoreGlobal,
|
||||
}
|
||||
payloadBytes, _ := json.Marshal(payload)
|
||||
|
||||
req, err := http.NewRequest("POST", url, strings.NewReader(string(payloadBytes)))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("server returned HTTP %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var wrapper struct {
|
||||
Code int `json:"code"`
|
||||
Data map[string]string `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &wrapper); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if wrapper.Code != 0 {
|
||||
return fmt.Errorf("server error code: %d", wrapper.Code)
|
||||
}
|
||||
|
||||
// update timestamp
|
||||
if ts, ok := wrapper.Data["latest.fetch.timestamp"]; ok {
|
||||
fmt.Sscanf(ts, "%d", &c.lastTime)
|
||||
delete(wrapper.Data, "latest.fetch.timestamp")
|
||||
}
|
||||
|
||||
if len(wrapper.Data) > 0 {
|
||||
// decrypt values
|
||||
changed := make(map[string]string)
|
||||
for k, v := range wrapper.Data {
|
||||
decrypted, err := decryptConfig(c.secret, v)
|
||||
if err != nil && c.globalSecret != "" {
|
||||
decrypted, err = decryptConfig(c.globalSecret, v)
|
||||
}
|
||||
if err != nil {
|
||||
decrypted = v
|
||||
}
|
||||
changed[k] = decrypted
|
||||
}
|
||||
|
||||
// update local configs
|
||||
c.mu.Lock()
|
||||
for k, v := range changed {
|
||||
c.configs[k] = v
|
||||
}
|
||||
c.mu.Unlock()
|
||||
|
||||
// notify listeners
|
||||
c.listener.Notify(changed)
|
||||
log.Printf("[ConfigClient] %d configs changed", len(changed))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Client) applyConfigs(data map[string]interface{}) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
for k, v := range data {
|
||||
// skip meta keys
|
||||
if strings.HasPrefix(k, "kenaito.") || k == "latest.fetch.timestamp" {
|
||||
continue
|
||||
}
|
||||
strVal, ok := v.(string)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
// decrypt if needed - try app secret first, then global secret
|
||||
decrypted, err := decryptConfig(c.secret, strVal)
|
||||
if err != nil && c.globalSecret != "" {
|
||||
decrypted, err = decryptConfig(c.globalSecret, strVal)
|
||||
}
|
||||
if err != nil {
|
||||
decrypted = strVal
|
||||
}
|
||||
c.configs[k] = decrypted
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) syncLoop() {
|
||||
ticker := time.NewTicker(time.Duration(c.config.SyncInterval) * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
if err := c.fetchChanged(); err != nil {
|
||||
log.Printf("[ConfigClient] sync error: %v", err)
|
||||
}
|
||||
case <-c.stopCh:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user