355 lines
11 KiB
Go
355 lines
11 KiB
Go
package handler
|
|
|
|
import (
|
|
"fmt"
|
|
"kenaito-conf/server/internal/dao"
|
|
"kenaito-conf/server/internal/logger"
|
|
"kenaito-conf/server/internal/model"
|
|
"kenaito-conf/server/internal/service"
|
|
"net"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
)
|
|
|
|
// getClientIP 获取客户端 IPv4 地址，将 IPv6 环回地址转为 127.0.0.1
|
|
func getClientIP(c *gin.Context) string {
|
|
ip := c.ClientIP()
|
|
if ip == "" {
|
|
return ""
|
|
}
|
|
parsed := net.ParseIP(ip)
|
|
if parsed == nil {
|
|
return ip
|
|
}
|
|
if parsed.IsLoopback() {
|
|
return "127.0.0.1"
|
|
}
|
|
if v4 := parsed.To4(); v4 != nil {
|
|
return v4.String()
|
|
}
|
|
return ip
|
|
}
|
|
|
|
type ConfigApiHandler struct {
|
|
parseService *service.ConfigParseService
|
|
}
|
|
|
|
func NewConfigApiHandler(parse *service.ConfigParseService) *ConfigApiHandler {
|
|
return &ConfigApiHandler{parseService: parse}
|
|
}
|
|
|
|
type FetchAllRequest struct {
|
|
AppName string `json:"appName" binding:"required"`
|
|
Env string `json:"env" binding:"required"`
|
|
Version string `json:"version"`
|
|
IgnoreGlobal bool `json:"ignoreGlobal"`
|
|
Token string `json:"token"`
|
|
}
|
|
|
|
func (h *ConfigApiHandler) FetchAllConfigs(c *gin.Context) {
|
|
var req FetchAllRequest
|
|
if err := c.ShouldBindJSON(&req); err != nil {
|
|
c.JSON(http.StatusBadRequest, model.WrapperResponse{Code: 400, Msg: "参数错误"})
|
|
return
|
|
}
|
|
|
|
version := req.Version
|
|
if version == "" {
|
|
version = "latest"
|
|
}
|
|
|
|
if req.AppName == "" || req.Env == "" {
|
|
c.JSON(http.StatusBadRequest, model.WrapperResponse{Code: 400, Msg: "参数缺失"})
|
|
return
|
|
}
|
|
|
|
// find app by code
|
|
var app model.Application
|
|
has, _ := dao.Engine.Where("code = ? AND enabled = 1", req.AppName).Get(&app)
|
|
if !has {
|
|
c.JSON(http.StatusOK, model.WrapperResponse{Code: 1001, Msg: "配置不存在"})
|
|
return
|
|
}
|
|
|
|
// check if app's group is enabled
|
|
if app.GroupId != nil {
|
|
var group model.BusinessGroup
|
|
has, _ := dao.Engine.ID(*app.GroupId).Get(&group)
|
|
if !has || !group.Enabled {
|
|
c.JSON(http.StatusForbidden, model.WrapperResponse{Code: 403, Msg: "业务组已禁用"})
|
|
return
|
|
}
|
|
}
|
|
|
|
// validate token if app has token configured
|
|
if app.Token != "" && app.Token != req.Token {
|
|
c.JSON(http.StatusForbidden, model.WrapperResponse{Code: 403, Msg: "Token 无效"})
|
|
return
|
|
}
|
|
|
|
// 游标上界取查询前的数据库时间，保证查询期间发生的更新会在下次增量中返回
|
|
upperBound, err := dao.Now()
|
|
if err != nil {
|
|
logger.SugaredLogger.Errorf("[FetchAllConfigs] get database time error: %v", err)
|
|
c.JSON(http.StatusInternalServerError, model.WrapperResponse{Code: 500, Msg: "服务暂时不可用"})
|
|
return
|
|
}
|
|
|
|
result := make(map[string]interface{})
|
|
var configIds []int
|
|
|
|
// load global configs
|
|
if !req.IgnoreGlobal && app.GroupId != nil {
|
|
var globalConfigs []model.AppConfig
|
|
dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND group_id = ? AND (app_id is null OR app_id = 0)",
|
|
req.Env, version, *app.GroupId).Find(&globalConfigs)
|
|
for _, cfg := range globalConfigs {
|
|
kvMap, err := h.parseService.ParseConfigToKVMap(&cfg)
|
|
if err != nil {
|
|
logger.SugaredLogger.Errorf("[FetchAllConfigs] parse config %d (%s) error: %v", cfg.Id, cfg.Name, err)
|
|
continue
|
|
}
|
|
configIds = append(configIds, cfg.Id)
|
|
for k, v := range kvMap {
|
|
result[k] = v
|
|
}
|
|
}
|
|
}
|
|
|
|
// load app configs
|
|
var appConfigs []model.AppConfig
|
|
dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND app_id = ?",
|
|
req.Env, version, app.Id).Find(&appConfigs)
|
|
for _, cfg := range appConfigs {
|
|
kvMap, err := h.parseService.ParseConfigToKVMap(&cfg)
|
|
if err != nil {
|
|
logger.SugaredLogger.Errorf("[FetchAllConfigs] parse config %d (%s) error: %v", cfg.Id, cfg.Name, err)
|
|
continue
|
|
}
|
|
configIds = append(configIds, cfg.Id)
|
|
for k, v := range kvMap {
|
|
result[k] = v
|
|
}
|
|
}
|
|
|
|
if len(result) == 0 {
|
|
c.JSON(http.StatusOK, model.WrapperResponse{Code: 1001, Msg: "配置不存在"})
|
|
return
|
|
}
|
|
|
|
result["latest.fetch.timestamp"] = strconv.FormatInt(upperBound.UnixMilli(), 10)
|
|
|
|
// record client instances asynchronously
|
|
go h.recordClientInstances(getClientIP(c), configIds, req.AppName, req.Env, "fetchAll")
|
|
|
|
c.JSON(http.StatusOK, result)
|
|
}
|
|
|
|
type FetchChangedRequest struct {
|
|
AppName string `json:"appName" binding:"required"`
|
|
Env string `json:"env" binding:"required"`
|
|
Version string `json:"version"`
|
|
LastTime int64 `json:"lastTime" binding:"required"`
|
|
IgnoreGlobal bool `json:"ignoreGlobal"`
|
|
Token string `json:"token"`
|
|
}
|
|
|
|
func (h *ConfigApiHandler) FetchChangedConfigs(c *gin.Context) {
|
|
var req FetchChangedRequest
|
|
if err := c.ShouldBindJSON(&req); err != nil {
|
|
c.JSON(http.StatusBadRequest, model.WrapperResponse{Code: 400, Msg: "参数错误"})
|
|
return
|
|
}
|
|
|
|
var app model.Application
|
|
has, _ := dao.Engine.Where("code = ? AND enabled = 1", req.AppName).Get(&app)
|
|
if !has {
|
|
c.JSON(http.StatusOK, model.WrapperResponse{Code: 1001, Msg: "应用不存在"})
|
|
return
|
|
}
|
|
|
|
// validate token if app has token configured
|
|
if app.Token != "" && app.Token != req.Token {
|
|
c.JSON(http.StatusForbidden, model.WrapperResponse{Code: 403, Msg: "Token 无效"})
|
|
return
|
|
}
|
|
|
|
version := req.Version
|
|
if version == "" {
|
|
version = "latest"
|
|
}
|
|
|
|
lastUpdateTime := time.UnixMilli(req.LastTime)
|
|
|
|
// 以数据库时间为本次查询的固定上界，避免查询与游标生成之间的竞态，
|
|
// 同时避免多实例应用服务器的时钟偏差。
|
|
upperBound, err := dao.Now()
|
|
if err != nil {
|
|
logger.SugaredLogger.Errorf("[FetchChangedConfigs] get database time error: %v", err)
|
|
c.JSON(http.StatusInternalServerError, model.WrapperResponse{Code: 500, Msg: "服务暂时不可用"})
|
|
return
|
|
}
|
|
|
|
result := make(map[string]string)
|
|
var configIds []int
|
|
|
|
// 查询已删除的配置（enabled=0 且 updated_at > lastTime），返回 deletedKeys
|
|
var deletedKeys []string
|
|
if !req.IgnoreGlobal && app.GroupId != nil {
|
|
var deletedGlobalConfigs []model.AppConfig
|
|
dao.Engine.Where("env = ? AND version = ? AND enabled = 0 AND group_id = ? AND (app_id is null OR app_id = 0) AND updated_at > ? AND updated_at <= ?",
|
|
req.Env, version, *app.GroupId, lastUpdateTime, upperBound).Find(&deletedGlobalConfigs)
|
|
for _, cfg := range deletedGlobalConfigs {
|
|
kvMap, err := h.parseService.ParseConfigToKVMap(&cfg)
|
|
if err != nil {
|
|
logger.SugaredLogger.Errorf("[FetchChangedConfigs] parse deleted config %d (%s) error: %v", cfg.Id, cfg.Name, err)
|
|
continue
|
|
}
|
|
for k := range kvMap {
|
|
deletedKeys = append(deletedKeys, k)
|
|
}
|
|
}
|
|
}
|
|
var deletedAppConfigs []model.AppConfig
|
|
dao.Engine.Where("env = ? AND version = ? AND enabled = 0 AND app_id = ? AND updated_at > ? AND updated_at <= ?",
|
|
req.Env, version, app.Id, lastUpdateTime, upperBound).Find(&deletedAppConfigs)
|
|
for _, cfg := range deletedAppConfigs {
|
|
kvMap, err := h.parseService.ParseConfigToKVMap(&cfg)
|
|
if err != nil {
|
|
logger.SugaredLogger.Errorf("[FetchChangedConfigs] parse deleted config %d (%s) error: %v", cfg.Id, cfg.Name, err)
|
|
continue
|
|
}
|
|
for k := range kvMap {
|
|
deletedKeys = append(deletedKeys, k)
|
|
}
|
|
}
|
|
if len(deletedKeys) > 0 {
|
|
result["deletedKeys"] = strings.Join(deletedKeys, ",")
|
|
}
|
|
|
|
// load global changed configs
|
|
if !req.IgnoreGlobal && app.GroupId != nil {
|
|
var globalConfigs []model.AppConfig
|
|
dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND group_id = ? AND (app_id is null OR app_id = 0) AND updated_at > ? AND updated_at <= ?",
|
|
req.Env, version, *app.GroupId, lastUpdateTime, upperBound).Find(&globalConfigs)
|
|
for _, cfg := range globalConfigs {
|
|
if h.addChangedConfigs(result, &cfg) {
|
|
configIds = append(configIds, cfg.Id)
|
|
}
|
|
}
|
|
}
|
|
|
|
// load app changed configs
|
|
var appConfigs1 []model.AppConfig
|
|
dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND app_id = ? AND updated_at > ? AND updated_at <= ?",
|
|
req.Env, version, app.Id, lastUpdateTime, upperBound).Find(&appConfigs1)
|
|
for _, cfg := range appConfigs1 {
|
|
if h.addChangedConfigs(result, &cfg) {
|
|
configIds = append(configIds, cfg.Id)
|
|
}
|
|
}
|
|
|
|
result["latest.fetch.timestamp"] = strconv.FormatInt(upperBound.UnixMilli(), 10)
|
|
|
|
// query all config ids for this app to record client instances
|
|
var allConfigIds []int
|
|
if !req.IgnoreGlobal && app.GroupId != nil {
|
|
var globalConfigs []model.AppConfig
|
|
dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND group_id = ? AND (app_id is null OR app_id = 0)",
|
|
req.Env, version, *app.GroupId).Find(&globalConfigs)
|
|
for _, cfg := range globalConfigs {
|
|
allConfigIds = append(allConfigIds, cfg.Id)
|
|
}
|
|
}
|
|
var appConfigs2 []model.AppConfig
|
|
dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND app_id = ?",
|
|
req.Env, version, app.Id).Find(&appConfigs2)
|
|
for _, cfg := range appConfigs2 {
|
|
allConfigIds = append(allConfigIds, cfg.Id)
|
|
}
|
|
|
|
// record client instances asynchronously
|
|
// only update last_fetch_at if there are actual changes
|
|
fetchType := ""
|
|
if len(configIds) > 0 {
|
|
fetchType = "fetchChanged"
|
|
}
|
|
go h.recordClientInstances(getClientIP(c), allConfigIds, req.AppName, req.Env, fetchType)
|
|
|
|
c.JSON(http.StatusOK, model.WrapperResponse{Code: 0, Data: result})
|
|
}
|
|
|
|
func (h *ConfigApiHandler) addChangedConfigs(result map[string]string, config *model.AppConfig) bool {
|
|
kvMap, err := h.parseService.ParseConfigToKVMap(config)
|
|
if err != nil {
|
|
logger.SugaredLogger.Errorf("[FetchChangedConfigs] parse config %d (%s) error: %v", config.Id, config.Name, err)
|
|
return false
|
|
}
|
|
for k, v := range kvMap {
|
|
result[k] = fmt.Sprintf("%v", v)
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (h *ConfigApiHandler) recordClientInstances(ip string, configIds []int, appName, env, fetchType string) {
|
|
if len(configIds) == 0 || ip == "" {
|
|
return
|
|
}
|
|
now := time.Now()
|
|
|
|
// 批量查询已存在的记录
|
|
var existingList []model.ClientInstance
|
|
dao.Engine.Where("app_name = ? AND env = ? AND ip = ?", appName, env, ip).
|
|
In("app_config_id", configIds).
|
|
Find(&existingList)
|
|
|
|
existingMap := make(map[int]*model.ClientInstance, len(existingList))
|
|
for i := range existingList {
|
|
existingMap[existingList[i].AppConfigId] = &existingList[i]
|
|
}
|
|
|
|
var toInsert []model.ClientInstance
|
|
for _, configId := range configIds {
|
|
if existing, ok := existingMap[configId]; ok {
|
|
existing.LastPingAt = &now
|
|
existing.UpdatedAt = &now
|
|
if fetchType != "" {
|
|
existing.LastFetchAt = &now
|
|
existing.LastFetchType = fetchType
|
|
dao.Engine.ID(existing.Id).Cols("last_ping_at", "last_fetch_at", "last_fetch_type", "updated_at").Update(existing)
|
|
} else {
|
|
dao.Engine.ID(existing.Id).Cols("last_ping_at", "updated_at").Update(existing)
|
|
}
|
|
} else {
|
|
instance := model.ClientInstance{
|
|
AppConfigId: configId,
|
|
AppName: appName,
|
|
Env: env,
|
|
Ip: ip,
|
|
LastPingAt: &now,
|
|
CreatedAt: &now,
|
|
UpdatedAt: &now,
|
|
}
|
|
if fetchType != "" {
|
|
instance.LastFetchAt = &now
|
|
instance.LastFetchType = fetchType
|
|
}
|
|
toInsert = append(toInsert, instance)
|
|
}
|
|
}
|
|
|
|
if len(toInsert) > 0 {
|
|
if _, err := dao.Engine.Insert(&toInsert); err != nil {
|
|
logger.SugaredLogger.Errorf("[recordClientInstances] batch insert error: %v", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// func (h *ConfigApiHandler) Ping(c *gin.Context) {
|
|
// c.String(http.StatusOK, "200")
|
|
// }
|