refactor: 客户端目录统一添加 kenaito-conf- 前缀
- go-client → kenaito-conf-go-client - python-client → kenaito-conf-python-client - python-client-demo → kenaito-conf-python-client-demo - nodejs-client → kenaito-conf-nodejs-client - nodejs-client-demo → kenaito-conf-nodejs-client-demo - 更新所有 README 和 package.json 中的路径引用
This commit is contained in:
@@ -0,0 +1,38 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
const cacheFile = "config-cache.json"
|
||||
|
||||
// Cache provides local file caching for configs
|
||||
type Cache struct {
|
||||
dir string
|
||||
}
|
||||
|
||||
func NewCache(dir string) *Cache {
|
||||
return &Cache{dir: dir}
|
||||
}
|
||||
|
||||
func (c *Cache) Write(data map[string]interface{}) error {
|
||||
path := filepath.Join(c.dir, cacheFile)
|
||||
bytes, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, bytes, 0644)
|
||||
}
|
||||
|
||||
func (c *Cache) Read() (map[string]interface{}, error) {
|
||||
path := filepath.Join(c.dir, cacheFile)
|
||||
bytes, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var data map[string]interface{}
|
||||
err = json.Unmarshal(bytes, &data)
|
||||
return data, err
|
||||
}
|
||||
@@ -0,0 +1,304 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"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", c.config.BaseURL)
|
||||
|
||||
payload := map[string]interface{}{
|
||||
"appName": c.config.AppName,
|
||||
"env": c.config.Env,
|
||||
"version": c.config.Version,
|
||||
"ignoreGlobal": c.config.IgnoreGlobal,
|
||||
"token": c.config.Token,
|
||||
}
|
||||
payloadBytes, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
req, err := http.NewRequest("POST", url, bytes.NewReader(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 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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package client
|
||||
|
||||
// Config holds the client configuration
|
||||
type Config struct {
|
||||
BaseURL string `json:"baseUrl"`
|
||||
AppName string `json:"appName"`
|
||||
Env string `json:"env"`
|
||||
Version string `json:"version"`
|
||||
Token string `json:"token"`
|
||||
SyncInterval int `json:"syncInterval"` // seconds
|
||||
IgnoreGlobal bool `json:"ignoreGlobal"`
|
||||
CacheDir string `json:"cacheDir"`
|
||||
}
|
||||
|
||||
// DefaultConfig returns a config with default values
|
||||
func DefaultConfig() *Config {
|
||||
return &Config{
|
||||
Version: "latest",
|
||||
SyncInterval: 60,
|
||||
CacheDir: ".",
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"crypto/aes"
|
||||
"crypto/cipher"
|
||||
"crypto/md5"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
const CryptPrefix = "{Cipher}"
|
||||
|
||||
func decryptConfig(secret string, data string) (string, error) {
|
||||
if !strings.HasPrefix(data, CryptPrefix) {
|
||||
return data, nil
|
||||
}
|
||||
encrypted := strings.TrimPrefix(data, CryptPrefix)
|
||||
return decryptWithAES(secret, encrypted)
|
||||
}
|
||||
|
||||
func decryptWithAES(key string, data string) (string, error) {
|
||||
secretKey := md5.Sum([]byte(key))
|
||||
aesKey := hex.EncodeToString(secretKey[:])[16:]
|
||||
|
||||
cipherText, err := base64.StdEncoding.DecodeString(data)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
block, err := aes.NewCipher([]byte(aesKey))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if len(cipherText)%aes.BlockSize != 0 {
|
||||
return "", fmt.Errorf("ciphertext is not a multiple of block size")
|
||||
}
|
||||
|
||||
plainText := make([]byte, len(cipherText))
|
||||
mode := cipher.NewCBCDecrypter(block, make([]byte, aes.BlockSize))
|
||||
mode.CryptBlocks(plainText, cipherText)
|
||||
|
||||
plainText, err = pkcs7Unpad(plainText, aes.BlockSize)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return string(plainText), nil
|
||||
}
|
||||
|
||||
func pkcs7Unpad(data []byte, blockSize int) ([]byte, error) {
|
||||
if len(data) == 0 {
|
||||
return nil, fmt.Errorf("empty data")
|
||||
}
|
||||
padding := int(data[len(data)-1])
|
||||
if padding > blockSize || padding == 0 {
|
||||
return nil, fmt.Errorf("invalid padding")
|
||||
}
|
||||
return data[:len(data)-padding], nil
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"kenaito-conf/go-client"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
func main() {
|
||||
// create config
|
||||
cfg := &client.Config{
|
||||
BaseURL: "http://localhost:8080",
|
||||
AppName: "kenaito-springboot-demo",
|
||||
Env: "daily",
|
||||
Version: "latest",
|
||||
//Token: "6f6966fe196572954b176c0e2a97bc399b050c6be31454b8d6b0f7f622b5d6f1",
|
||||
SyncInterval: 30, // sync every 30 seconds
|
||||
CacheDir: ".",
|
||||
}
|
||||
|
||||
// create client
|
||||
c := client.New(cfg)
|
||||
|
||||
// register change handler
|
||||
c.OnChange(func(changed map[string]string) {
|
||||
fmt.Println("Config changed:")
|
||||
for k, v := range changed {
|
||||
fmt.Printf(" %s = %s\n", k, v)
|
||||
}
|
||||
})
|
||||
|
||||
// start client
|
||||
if err := c.Start(); err != nil {
|
||||
log.Fatalf("Failed to start config client: %v", err)
|
||||
}
|
||||
defer c.Stop()
|
||||
|
||||
// read configs
|
||||
fmt.Println("Current configs:")
|
||||
for k, v := range c.GetAll() {
|
||||
fmt.Printf(" %s = %s\n", k, v)
|
||||
}
|
||||
|
||||
// get specific config
|
||||
dbHost := c.Get("database.host")
|
||||
fmt.Printf("\ndatabase.host = %s\n", dbHost)
|
||||
|
||||
// keep running until signal
|
||||
fmt.Println("\nListening for config changes... (press Ctrl+C to stop)")
|
||||
sigCh := make(chan os.Signal, 1)
|
||||
signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
|
||||
<-sigCh
|
||||
fmt.Println("\nShutting down...")
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
module kenaito-conf/go-client
|
||||
|
||||
go 1.25.0
|
||||
@@ -0,0 +1,39 @@
|
||||
package client
|
||||
|
||||
import "sync"
|
||||
|
||||
// ChangeHandler is called when configs change
|
||||
type ChangeHandler func(changed map[string]string)
|
||||
|
||||
// Listener manages config change callbacks
|
||||
type Listener struct {
|
||||
mu sync.RWMutex
|
||||
handlers []ChangeHandler
|
||||
}
|
||||
|
||||
func NewListener() *Listener {
|
||||
return &Listener{
|
||||
handlers: make([]ChangeHandler, 0),
|
||||
}
|
||||
}
|
||||
|
||||
func (l *Listener) OnChange(handler ChangeHandler) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
l.handlers = append(l.handlers, handler)
|
||||
}
|
||||
|
||||
func (l *Listener) Notify(changed map[string]string) {
|
||||
l.mu.RLock()
|
||||
defer l.mu.RUnlock()
|
||||
for _, h := range l.handlers {
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
// log and continue
|
||||
}
|
||||
}()
|
||||
h(changed)
|
||||
}()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user