robin/waf-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

1.5.0

Browse Source
This commit is contained in:
robin
2026-03-13 14:25:13 +08:00
parent a25a474d6a
commit afbaaa869c
95 changed files with 4591 additions and 2578 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
EdgeAPI/build/build.sh
View File

@@ -128,13 +128,11 @@ function build() {
# copy files
echo "copying ..."
if [ ! -d "$DIST" ]; then
mkdir "$DIST"
mkdir "$DIST"/bin
mkdir "$DIST"/configs
mkdir "$DIST"/logs
mkdir "$DIST"/data
fi
rm -rf "$DIST"
mkdir -p "$DIST"/bin
mkdir -p "$DIST"/configs
mkdir -p "$DIST"/logs
mkdir -p "$DIST"/data
cp "$ROOT"/configs/api.template.yaml "$DIST"/configs/
cp "$ROOT"/configs/db.template.yaml "$DIST"/configs/
# 复制 EdgeCommon 的配置文件（如果存在）
@@ -148,42 +146,6 @@ function build() {
rm -f "$DIST"/deploy/.gitignore
cp -R "$ROOT"/installers "$DIST"/
# copy fluent-bit templates and local packages from repo root
FLUENT_ROOT="$ROOT/../../deploy/fluent-bit"
FLUENT_DIST="$DIST/deploy/fluent-bit"
if [ -d "$FLUENT_ROOT" ]; then
rm -rf "$FLUENT_DIST"
mkdir -p "$FLUENT_DIST"
FLUENT_FILES=(
"fluent-bit.conf"
"fluent-bit-dns.conf"
"fluent-bit-https.conf"
"fluent-bit-dns-https.conf"
"fluent-bit-windows.conf"
"fluent-bit-windows-https.conf"
"parsers.conf"
"clickhouse-upstream.conf"
"clickhouse-upstream-windows.conf"
"logrotate.conf"
"README.md"
)
for file in "${FLUENT_FILES[@]}"; do
if [ -f "$FLUENT_ROOT/$file" ]; then
cp "$FLUENT_ROOT/$file" "$FLUENT_DIST/"
fi
done
if [ -d "$FLUENT_ROOT/packages" ]; then
cp -R "$FLUENT_ROOT/packages" "$FLUENT_DIST/"
fi
# remove local runtime artifacts if present
rm -f "$FLUENT_DIST/.gitignore"
rm -f "$FLUENT_DIST"/logs.db*
rm -rf "$FLUENT_DIST/storage"
fi
# building edge installer
echo "building node installer ..."
architects=("amd64")

176
EdgeAPI/internal/clickhouse/httpdns_access_logs_store.go
View File

@@ -47,6 +47,33 @@ type HTTPDNSAccessLogListFilter struct {
Size int64
}
type HTTPDNSAccessLogHourlyStat struct {
Hour string
CountRequests int64
}
type HTTPDNSAccessLogDailyStat struct {
Day string
CountRequests int64
}
type HTTPDNSAccessLogTopAppStat struct {
AppId string
AppName string
CountRequests int64
}
type HTTPDNSAccessLogTopDomainStat struct {
Domain string
CountRequests int64
}
type HTTPDNSAccessLogTopNodeStat struct {
ClusterId uint32
NodeId uint32
CountRequests int64
}
type HTTPDNSAccessLogsStore struct {
client *Client
}
@@ -176,6 +203,155 @@ func (s *HTTPDNSAccessLogsStore) List(ctx context.Context, f HTTPDNSAccessLogLis
return result, nil
}
func (s *HTTPDNSAccessLogsStore) FindHourlyStats(ctx context.Context, hourFrom string, hourTo string) ([]*HTTPDNSAccessLogHourlyStat, error) {
if !s.client.IsConfigured() {
return nil, fmt.Errorf("clickhouse: not configured")
}
query := fmt.Sprintf(
"SELECT formatDateTime(toDateTime(created_at), '%%Y%%m%%d%%H') AS hour, count() AS count_requests FROM %s WHERE formatDateTime(toDateTime(created_at), '%%Y%%m%%d%%H') BETWEEN '%s' AND '%s' GROUP BY hour ORDER BY hour ASC",
s.tableName(),
escapeString(hourFrom),
escapeString(hourTo),
)
rows := []map[string]interface{}{}
if err := s.client.Query(ctx, query, &rows); err != nil {
return nil, err
}
result := make([]*HTTPDNSAccessLogHourlyStat, 0, len(rows))
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogHourlyStat{
Hour: toString(row["hour"]),
CountRequests: toInt64(row["count_requests"]),
})
}
return result, nil
}
func (s *HTTPDNSAccessLogsStore) FindDailyStats(ctx context.Context, dayFrom string, dayTo string) ([]*HTTPDNSAccessLogDailyStat, error) {
if !s.client.IsConfigured() {
return nil, fmt.Errorf("clickhouse: not configured")
}
query := fmt.Sprintf(
"SELECT day, count() AS count_requests FROM %s WHERE day BETWEEN '%s' AND '%s' GROUP BY day ORDER BY day ASC",
s.tableName(),
escapeString(dayFrom),
escapeString(dayTo),
)
rows := []map[string]interface{}{}
if err := s.client.Query(ctx, query, &rows); err != nil {
return nil, err
}
result := make([]*HTTPDNSAccessLogDailyStat, 0, len(rows))
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogDailyStat{
Day: toString(row["day"]),
CountRequests: toInt64(row["count_requests"]),
})
}
return result, nil
}
func (s *HTTPDNSAccessLogsStore) ListTopApps(ctx context.Context, dayFrom string, dayTo string, limit int64) ([]*HTTPDNSAccessLogTopAppStat, error) {
if !s.client.IsConfigured() {
return nil, fmt.Errorf("clickhouse: not configured")
}
if limit <= 0 {
limit = 10
}
query := fmt.Sprintf(
"SELECT app_id, max(app_name) AS app_name, count() AS count_requests FROM %s WHERE day BETWEEN '%s' AND '%s' GROUP BY app_id ORDER BY count_requests DESC LIMIT %d",
s.tableName(),
escapeString(dayFrom),
escapeString(dayTo),
limit,
)
rows := []map[string]interface{}{}
if err := s.client.Query(ctx, query, &rows); err != nil {
return nil, err
}
result := make([]*HTTPDNSAccessLogTopAppStat, 0, len(rows))
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogTopAppStat{
AppId: toString(row["app_id"]),
AppName: toString(row["app_name"]),
CountRequests: toInt64(row["count_requests"]),
})
}
return result, nil
}
func (s *HTTPDNSAccessLogsStore) ListTopDomains(ctx context.Context, dayFrom string, dayTo string, limit int64) ([]*HTTPDNSAccessLogTopDomainStat, error) {
if !s.client.IsConfigured() {
return nil, fmt.Errorf("clickhouse: not configured")
}
if limit <= 0 {
limit = 10
}
query := fmt.Sprintf(
"SELECT domain, count() AS count_requests FROM %s WHERE day BETWEEN '%s' AND '%s' AND domain != '' GROUP BY domain ORDER BY count_requests DESC LIMIT %d",
s.tableName(),
escapeString(dayFrom),
escapeString(dayTo),
limit,
)
rows := []map[string]interface{}{}
if err := s.client.Query(ctx, query, &rows); err != nil {
return nil, err
}
result := make([]*HTTPDNSAccessLogTopDomainStat, 0, len(rows))
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogTopDomainStat{
Domain: toString(row["domain"]),
CountRequests: toInt64(row["count_requests"]),
})
}
return result, nil
}
func (s *HTTPDNSAccessLogsStore) ListTopNodes(ctx context.Context, dayFrom string, dayTo string, limit int64) ([]*HTTPDNSAccessLogTopNodeStat, error) {
if !s.client.IsConfigured() {
return nil, fmt.Errorf("clickhouse: not configured")
}
if limit <= 0 {
limit = 10
}
query := fmt.Sprintf(
"SELECT min(cluster_id) AS cluster_id, node_id, count() AS count_requests FROM %s WHERE day BETWEEN '%s' AND '%s' GROUP BY node_id ORDER BY count_requests DESC LIMIT %d",
s.tableName(),
escapeString(dayFrom),
escapeString(dayTo),
limit,
)
rows := []map[string]interface{}{}
if err := s.client.Query(ctx, query, &rows); err != nil {
return nil, err
}
result := make([]*HTTPDNSAccessLogTopNodeStat, 0, len(rows))
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogTopNodeStat{
ClusterId: uint32(toInt64(row["cluster_id"])),
NodeId: uint32(toInt64(row["node_id"])),
CountRequests: toInt64(row["count_requests"]),
})
}
return result, nil
}
func HTTPDNSRowToPB(row *HTTPDNSAccessLogRow) *pb.HTTPDNSAccessLog {
if row == nil {
return nil

4
EdgeAPI/internal/const/const.go
View File

@@ -1,7 +1,7 @@
package teaconst
const (
Version = "1.4.9" //1.3.9
Version = "1.5.0" //1.3.9
ProductName = "Edge API"
ProcessName = "edge-api"
@@ -17,6 +17,6 @@ const (
// 其他节点版本号，用来检测是否有需要升级的节点
NodeVersion = "1.4.9" //1.3.8.2
NodeVersion = "1.5.0" //1.3.8.2
)

6
EdgeAPI/internal/const/const_plus.go
View File

@@ -4,9 +4,9 @@
package teaconst
const (
DNSNodeVersion = "1.4.9" //1.3.8.2
UserNodeVersion = "1.4.9" //1.3.8.2
ReportNodeVersion = "0.1.5"
DNSNodeVersion = "1.5.0" //1.3.8.2
UserNodeVersion = "1.5.0" //1.3.8.2
ReportNodeVersion = "1.5.0"
DefaultMaxNodes int32 = 50
)

138
EdgeAPI/internal/db/models/httpdns_access_log_dao_stat.go Normal file
View File

@@ -0,0 +1,138 @@
package models
import (
"github.com/iwind/TeaGo/dbs"
)
type HTTPDNSAccessLogHourlyStat struct {
Hour string `field:"hour"`
CountRequests int64 `field:"countRequests"`
}
type HTTPDNSAccessLogDailyStat struct {
Day string `field:"day"`
CountRequests int64 `field:"countRequests"`
}
type HTTPDNSAccessLogTopAppStat struct {
AppId string `field:"appId"`
AppName string `field:"appName"`
CountRequests int64 `field:"countRequests"`
}
type HTTPDNSAccessLogTopDomainStat struct {
Domain string `field:"domain"`
CountRequests int64 `field:"countRequests"`
}
type HTTPDNSAccessLogTopNodeStat struct {
ClusterId uint32 `field:"clusterId"`
NodeId uint32 `field:"nodeId"`
CountRequests int64 `field:"countRequests"`
}
func (this *HTTPDNSAccessLogDAO) FindHourlyStats(tx *dbs.Tx, hourFrom string, hourTo string) (result []*HTTPDNSAccessLogHourlyStat, err error) {
rows, _, err := this.Query(tx).
Result("CONCAT(day, LPAD(HOUR(FROM_UNIXTIME(createdAt)),2,'0')) AS hour", "COUNT(*) AS countRequests").
Where("CONCAT(day, LPAD(HOUR(FROM_UNIXTIME(createdAt)),2,'0')) BETWEEN :hourFrom AND :hourTo").
Param("hourFrom", hourFrom).
Param("hourTo", hourTo).
Group("hour").
FindOnes()
if err != nil {
return nil, err
}
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogHourlyStat{
Hour: row.GetString("hour"),
CountRequests: row.GetInt64("countRequests"),
})
}
return
}
func (this *HTTPDNSAccessLogDAO) FindDailyStats(tx *dbs.Tx, dayFrom string, dayTo string) (result []*HTTPDNSAccessLogDailyStat, err error) {
rows, _, err := this.Query(tx).
Result("day", "COUNT(*) AS countRequests").
Between("day", dayFrom, dayTo).
Group("day").
FindOnes()
if err != nil {
return nil, err
}
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogDailyStat{
Day: row.GetString("day"),
CountRequests: row.GetInt64("countRequests"),
})
}
return
}
func (this *HTTPDNSAccessLogDAO) ListTopApps(tx *dbs.Tx, dayFrom string, dayTo string, limit int64) (result []*HTTPDNSAccessLogTopAppStat, err error) {
rows, _, err := this.Query(tx).
Result("appId", "MAX(appName) AS appName", "COUNT(*) AS countRequests").
Between("day", dayFrom, dayTo).
Group("appId").
Desc("countRequests").
Limit(limit).
FindOnes()
if err != nil {
return nil, err
}
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogTopAppStat{
AppId: row.GetString("appId"),
AppName: row.GetString("appName"),
CountRequests: row.GetInt64("countRequests"),
})
}
return
}
func (this *HTTPDNSAccessLogDAO) ListTopDomains(tx *dbs.Tx, dayFrom string, dayTo string, limit int64) (result []*HTTPDNSAccessLogTopDomainStat, err error) {
rows, _, err := this.Query(tx).
Result("domain", "COUNT(*) AS countRequests").
Between("day", dayFrom, dayTo).
Where("domain != ''").
Group("domain").
Desc("countRequests").
Limit(limit).
FindOnes()
if err != nil {
return nil, err
}
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogTopDomainStat{
Domain: row.GetString("domain"),
CountRequests: row.GetInt64("countRequests"),
})
}
return
}
func (this *HTTPDNSAccessLogDAO) ListTopNodes(tx *dbs.Tx, dayFrom string, dayTo string, limit int64) (result []*HTTPDNSAccessLogTopNodeStat, err error) {
rows, _, err := this.Query(tx).
Result("MIN(clusterId) AS clusterId", "nodeId", "COUNT(*) AS countRequests").
Between("day", dayFrom, dayTo).
Group("nodeId").
Desc("countRequests").
Limit(limit).
FindOnes()
if err != nil {
return nil, err
}
for _, row := range rows {
result = append(result, &HTTPDNSAccessLogTopNodeStat{
ClusterId: row.GetUint32("clusterId"),
NodeId: row.GetUint32("nodeId"),
CountRequests: row.GetInt64("countRequests"),
})
}
return
}

8
EdgeAPI/internal/db/models/httpdns_domain_dao.go
View File

@@ -113,3 +113,11 @@ func (this *HTTPDNSDomainDAO) ListEnabledDomainsWithAppId(tx *dbs.Tx, appDbId in
_, err = query.Slice(&result).FindAll()
return
}
func (this *HTTPDNSDomainDAO) CountEnabledDomains(tx *dbs.Tx, appDbId int64) (int64, error) {
query := this.Query(tx).State(HTTPDNSDomainStateEnabled)
if appDbId > 0 {
query = query.Attr("appId", appDbId)
}
return query.Count()
}

35
EdgeAPI/internal/db/models/node_value_dao.go
View File

@@ -221,6 +221,41 @@ func (this *NodeValueDAO) ListValuesForNSNodes(tx *dbs.Tx, item string, key stri
return
}
// ListValuesForHTTPDNSNodes 列出HTTPDNS节点相关的平均数据
func (this *NodeValueDAO) ListValuesForHTTPDNSNodes(tx *dbs.Tx, item string, key string, timeRange nodeconfigs.NodeValueRange) (result []*NodeValue, err error) {
query := this.Query(tx).
Attr("role", "httpdns").
Attr("item", item).
Result("AVG(JSON_EXTRACT(value, '$." + key + "')) AS value, MIN(createdAt) AS createdAt")
switch timeRange {
// TODO 支持更多的时间范围
case nodeconfigs.NodeValueRangeMinute:
fromMinute := timeutil.FormatTime("YmdHi", time.Now().Unix()-3600) // 一个小时之前的
query.Gte("minute", fromMinute)
query.Result("minute")
query.Group("minute")
default:
err = errors.New("invalid 'range' value: '" + timeRange + "'")
return
}
_, err = query.Slice(&result).
FindAll()
if err != nil {
return nil, err
}
for _, nodeValue := range result {
nodeValue.Value, _ = json.Marshal(maps.Map{
key: types.Float32(string(nodeValue.Value)),
})
}
return
}
// SumAllNodeValues 计算所有节点的某项参数值
func (this *NodeValueDAO) SumAllNodeValues(tx *dbs.Tx, role string, item nodeconfigs.NodeValueItem, param string, duration int32, durationUnit nodeconfigs.NodeValueDurationUnit) (total float64, avg float64, max float64, err error) {
if duration <= 0 {

889
EdgeAPI/internal/installers/fluent_bit.go
View File

@@ -1,889 +0,0 @@
package installers
import (
"crypto/sha256"
"encoding/json"
"errors"
"fmt"
"net/url"
"os"
slashpath "path"
"path/filepath"
"sort"
"strconv"
"strings"
"time"
teaconst "github.com/TeaOSLab/EdgeAPI/internal/const"
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/systemconfigs"
"github.com/iwind/TeaGo/Tea"
)
const (
fluentBitConfigDir = "/etc/fluent-bit"
fluentBitStorageDir = "/var/lib/fluent-bit/storage"
fluentBitMainConfigFile = "/etc/fluent-bit/fluent-bit.conf"
fluentBitParsersFile = "/etc/fluent-bit/parsers.conf"
fluentBitManagedMetaFile = "/etc/fluent-bit/.edge-managed.json"
fluentBitManagedEnvFile = "/etc/fluent-bit/.edge-managed.env"
fluentBitDropInDir = "/etc/systemd/system/fluent-bit.service.d"
fluentBitDropInFile = "/etc/systemd/system/fluent-bit.service.d/edge-managed.conf"
fluentBitServiceName = "fluent-bit"
fluentBitDefaultBinPath = "/opt/fluent-bit/bin/fluent-bit"
fluentBitLocalPackagesRoot = "packages"
fluentBitHTTPPathPattern = "/var/log/edge/edge-node/*.log"
fluentBitDNSPathPattern = "/var/log/edge/edge-dns/*.log"
fluentBitHTTPDNSPathPattern = "/var/log/edge/edge-httpdns/*.log"
fluentBitManagedMarker = "managed-by-edgeapi"
fluentBitRoleNode = "node"
fluentBitRoleDNS = "dns"
fluentBitRoleHTTPDNS = "httpdns"
)
var errFluentBitLocalPackageNotFound = errors.New("fluent-bit local package not found")
var fluentBitPackageFileMapping = map[string]string{
"ubuntu22.04-amd64": "fluent-bit_4.2.2_amd64.deb",
"ubuntu22.04-arm64": "fluent-bit_4.2.2_arm64.deb",
"amzn2023-amd64": "fluent-bit-4.2.2-1.x86_64.rpm",
"amzn2023-arm64": "fluent-bit-4.2.2-1.aarch64.rpm",
}
type fluentBitManagedMeta struct {
Roles []string `json:"roles"`
Hash string `json:"hash"`
UpdatedAt int64 `json:"updatedAt"`
SourceVersion string `json:"sourceVersion"`
}
type fluentBitDesiredConfig struct {
Roles []string
ClickHouse *systemconfigs.ClickHouseSetting
HTTPPathPattern string
DNSPathPattern string
HTTPDNSPathPattern string
}
// SetupFluentBit 安装并托管 Fluent Bit 配置（离线包 + 平台渲染配置）。
func (this *BaseInstaller) SetupFluentBit(role nodeconfigs.NodeRole) error {
if this.client == nil {
return errors.New("ssh client is nil")
}
uname := this.uname()
if !strings.Contains(uname, "Linux") {
return nil
}
tempDir := strings.TrimRight(this.client.UserHome(), "/") + "/.edge-fluent-bit"
_, _, _ = this.client.Exec("mkdir -p " + shQuote(tempDir))
defer func() {
_, _, _ = this.client.Exec("rm -rf " + shQuote(tempDir))
}()
if err := this.ensureFluentBitInstalled(tempDir); err != nil {
return err
}
_, stderr, err := this.client.Exec("mkdir -p " + shQuote(fluentBitConfigDir) + " " + shQuote(fluentBitStorageDir))
if err != nil {
return fmt.Errorf("prepare fluent-bit directories failed: %w, stderr: %s", err, stderr)
}
parserContent, err := this.readLocalParsersContent()
if err != nil {
return err
}
existingMeta, err := this.readManagedMeta()
if err != nil {
return err
}
mergedRoles, err := mergeManagedRoles(existingMeta, role)
if err != nil {
return err
}
desiredConfig, err := this.buildDesiredFluentBitConfig(mergedRoles)
if err != nil {
return err
}
configChanged, err := this.applyManagedConfig(tempDir, desiredConfig, parserContent, existingMeta)
if err != nil {
return err
}
binPath, err := this.lookupFluentBitBinPath()
if err != nil {
return err
}
if err := this.ensureFluentBitService(tempDir, binPath, configChanged); err != nil {
return err
}
return nil
}
func (this *BaseInstaller) ensureFluentBitInstalled(tempDir string) error {
binPath, _ := this.lookupFluentBitBinPath()
if binPath != "" {
return nil
}
platformKey, packageName, arch, err := this.detectRemotePlatformAndPackage()
if err != nil {
return fmt.Errorf("detect fluent-bit platform failed: %w", err)
}
if err := this.installFluentBitFromLocalPackage(tempDir, arch, packageName); err != nil {
if errors.Is(err, errFluentBitLocalPackageNotFound) {
expectedPath := filepath.Join("deploy", "fluent-bit", fluentBitLocalPackagesRoot, "linux-"+arch, packageName)
return fmt.Errorf("install fluent-bit failed: local package missing for platform '%s', expected '%s'", platformKey, expectedPath)
}
return fmt.Errorf("install fluent-bit from local package failed: %w", err)
}
binPath, err = this.lookupFluentBitBinPath()
if err != nil {
return err
}
if binPath == "" {
return errors.New("fluent-bit binary not found after local package install")
}
_, stderr, err := this.client.Exec(binPath + " --version")
if err != nil {
return fmt.Errorf("verify fluent-bit version failed: %w, stderr: %s", err, stderr)
}
return nil
}
func (this *BaseInstaller) installFluentBitFromLocalPackage(tempDir string, arch string, packageName string) error {
packageDir := filepath.Join(Tea.Root, "deploy", "fluent-bit", fluentBitLocalPackagesRoot, "linux-"+arch)
localPackagePath := filepath.Join(packageDir, packageName)
if _, err := os.Stat(localPackagePath); err != nil {
if os.IsNotExist(err) {
return errFluentBitLocalPackageNotFound
}
return fmt.Errorf("check local package failed: %w", err)
}
remotePackagePath := tempDir + "/" + filepath.Base(localPackagePath)
if err := this.client.Copy(localPackagePath, remotePackagePath, 0644); err != nil {
return fmt.Errorf("upload local package failed: %w", err)
}
var installCmd string
lowerName := strings.ToLower(localPackagePath)
switch {
case strings.HasSuffix(lowerName, ".deb"):
installCmd = "dpkg -i " + shQuote(remotePackagePath)
case strings.HasSuffix(lowerName, ".rpm"):
installCmd = "rpm -Uvh --force " + shQuote(remotePackagePath) + " || rpm -ivh --force " + shQuote(remotePackagePath)
case strings.HasSuffix(lowerName, ".tar.gz") || strings.HasSuffix(lowerName, ".tgz"):
extractDir := tempDir + "/extract"
installCmd = "rm -rf " + shQuote(extractDir) + "; mkdir -p " + shQuote(extractDir) + "; tar -xzf " + shQuote(remotePackagePath) + " -C " + shQuote(extractDir) + "; " +
"bin=$(find " + shQuote(extractDir) + " -type f -name fluent-bit | head -n 1); " +
"if [ -z \"$bin\" ]; then exit 3; fi; " +
"mkdir -p /opt/fluent-bit/bin /usr/local/bin; " +
"install -m 0755 \"$bin\" /opt/fluent-bit/bin/fluent-bit; " +
"ln -sf /opt/fluent-bit/bin/fluent-bit /usr/local/bin/fluent-bit"
default:
return fmt.Errorf("unsupported local package format: %s", packageName)
}
_, stderr, err := this.client.Exec(installCmd)
if err != nil {
return fmt.Errorf("install fluent-bit local package '%s' failed: %w, stderr: %s", filepath.Base(localPackagePath), err, stderr)
}
return nil
}
func (this *BaseInstaller) detectRemotePlatformAndPackage() (platformKey string, packageName string, arch string, err error) {
arch, err = this.detectRemoteLinuxArch()
if err != nil {
return "", "", "", err
}
releaseData, stderr, err := this.client.Exec("cat /etc/os-release")
if err != nil {
return "", "", "", fmt.Errorf("read /etc/os-release failed: %w, stderr: %s", err, stderr)
}
if strings.TrimSpace(releaseData) == "" {
return "", "", "", errors.New("/etc/os-release is empty")
}
releaseMap := parseOSRelease(releaseData)
id := strings.ToLower(strings.TrimSpace(releaseMap["ID"]))
versionID := strings.TrimSpace(releaseMap["VERSION_ID"])
var distro string
switch {
case id == "ubuntu" && strings.HasPrefix(versionID, "22.04"):
distro = "ubuntu22.04"
case id == "amzn" && strings.HasPrefix(versionID, "2023"):
distro = "amzn2023"
default:
return "", "", "", fmt.Errorf("unsupported linux platform id='%s' version='%s'", id, versionID)
}
platformKey = distro + "-" + arch
packageName, ok := fluentBitPackageFileMapping[platformKey]
if !ok {
return "", "", "", fmt.Errorf("no local package mapping for platform '%s'", platformKey)
}
return platformKey, packageName, arch, nil
}
func parseOSRelease(content string) map[string]string {
result := map[string]string{}
lines := strings.Split(content, "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" || strings.HasPrefix(line, "#") || !strings.Contains(line, "=") {
continue
}
parts := strings.SplitN(line, "=", 2)
key := strings.TrimSpace(parts[0])
value := strings.TrimSpace(parts[1])
value = strings.Trim(value, "\"")
result[key] = value
}
return result
}
func (this *BaseInstaller) detectRemoteLinuxArch() (string, error) {
stdout, stderr, err := this.client.Exec("uname -m")
if err != nil {
return "", fmt.Errorf("detect remote arch failed: %w, stderr: %s", err, stderr)
}
arch := strings.ToLower(strings.TrimSpace(stdout))
switch arch {
case "x86_64", "amd64":
return "amd64", nil
case "aarch64", "arm64":
return "arm64", nil
default:
return arch, nil
}
}
func (this *BaseInstaller) lookupFluentBitBinPath() (string, error) {
stdout, stderr, err := this.client.Exec("if command -v fluent-bit >/dev/null 2>&1; then command -v fluent-bit; elif [ -x " + fluentBitDefaultBinPath + " ]; then echo " + fluentBitDefaultBinPath + "; fi")
if err != nil {
return "", fmt.Errorf("lookup fluent-bit binary failed: %w, stderr: %s", err, stderr)
}
return strings.TrimSpace(stdout), nil
}
func (this *BaseInstaller) readLocalParsersContent() (string, error) {
parsersPath := filepath.Join(Tea.Root, "deploy", "fluent-bit", "parsers.conf")
data, err := os.ReadFile(parsersPath)
if err != nil {
return "", fmt.Errorf("read local parsers config failed: %w", err)
}
return string(data), nil
}
func (this *BaseInstaller) readManagedMeta() (*fluentBitManagedMeta, error) {
exists, err := this.remoteFileExists(fluentBitManagedMetaFile)
if err != nil {
return nil, err
}
if !exists {
return nil, nil
}
content, stderr, err := this.client.Exec("cat " + shQuote(fluentBitManagedMetaFile))
if err != nil {
return nil, fmt.Errorf("read fluent-bit managed metadata failed: %w, stderr: %s", err, stderr)
}
if strings.TrimSpace(content) == "" {
return nil, nil
}
meta := &fluentBitManagedMeta{}
if err := json.Unmarshal([]byte(content), meta); err != nil {
return nil, fmt.Errorf("decode fluent-bit managed metadata failed: %w", err)
}
meta.Roles = normalizeRoles(meta.Roles)
return meta, nil
}
func mergeManagedRoles(meta *fluentBitManagedMeta, role nodeconfigs.NodeRole) ([]string, error) {
roleName, err := mapNodeRole(role)
if err != nil {
return nil, err
}
roleSet := map[string]struct{}{}
if meta != nil {
for _, r := range normalizeRoles(meta.Roles) {
roleSet[r] = struct{}{}
}
}
roleSet[roleName] = struct{}{}
roles := make([]string, 0, len(roleSet))
for r := range roleSet {
roles = append(roles, r)
}
sort.Strings(roles)
return roles, nil
}
func mapNodeRole(role nodeconfigs.NodeRole) (string, error) {
switch role {
case nodeconfigs.NodeRoleNode:
return fluentBitRoleNode, nil
case nodeconfigs.NodeRoleDNS:
return fluentBitRoleDNS, nil
case nodeconfigs.NodeRoleHTTPDNS:
return fluentBitRoleHTTPDNS, nil
default:
return "", fmt.Errorf("unsupported fluent-bit role '%s'", role)
}
}
func normalizeRoles(rawRoles []string) []string {
roleSet := map[string]struct{}{}
for _, role := range rawRoles {
role = strings.ToLower(strings.TrimSpace(role))
if role != fluentBitRoleNode && role != fluentBitRoleDNS && role != fluentBitRoleHTTPDNS {
continue
}
roleSet[role] = struct{}{}
}
roles := make([]string, 0, len(roleSet))
for role := range roleSet {
roles = append(roles, role)
}
sort.Strings(roles)
return roles
}
func hasRole(roles []string, role string) bool {
for _, one := range roles {
if one == role {
return true
}
}
return false
}
func (this *BaseInstaller) buildDesiredFluentBitConfig(roles []string) (*fluentBitDesiredConfig, error) {
if len(roles) == 0 {
return nil, errors.New("fluent-bit roles should not be empty")
}
ch, err := models.SharedSysSettingDAO.ReadClickHouseConfig(nil)
if err != nil {
return nil, fmt.Errorf("read clickhouse setting failed: %w", err)
}
if ch == nil {
ch = &systemconfigs.ClickHouseSetting{}
}
if strings.TrimSpace(ch.Host) == "" {
ch.Host = "127.0.0.1"
}
ch.Scheme = strings.ToLower(strings.TrimSpace(ch.Scheme))
if ch.Scheme == "" {
ch.Scheme = "https"
}
if ch.Scheme != "http" && ch.Scheme != "https" {
return nil, fmt.Errorf("unsupported clickhouse scheme '%s'", ch.Scheme)
}
if ch.Port <= 0 {
if ch.Scheme == "https" {
ch.Port = 8443
} else {
ch.Port = 8443
}
}
if strings.TrimSpace(ch.Database) == "" {
ch.Database = "default"
}
if strings.TrimSpace(ch.User) == "" {
ch.User = "default"
}
// 当前平台策略：后台固定跳过 ClickHouse TLS 证书校验，不暴露 ServerName 配置。
ch.TLSSkipVerify = true
ch.TLSServerName = ""
httpPathPattern := fluentBitHTTPPathPattern
dnsPathPattern := fluentBitDNSPathPattern
httpdnsPathPattern := fluentBitHTTPDNSPathPattern
publicPolicyPath, err := this.readPublicAccessLogPolicyPath()
if err != nil {
return nil, err
}
policyDir := dirFromPolicyPath(publicPolicyPath)
if policyDir != "" {
pattern := strings.TrimRight(policyDir, "/") + "/*.log"
httpPathPattern = pattern
dnsPathPattern = pattern
httpdnsPathPattern = pattern
}
return &fluentBitDesiredConfig{
Roles: normalizeRoles(roles),
ClickHouse: ch,
HTTPPathPattern: httpPathPattern,
DNSPathPattern: dnsPathPattern,
HTTPDNSPathPattern: httpdnsPathPattern,
}, nil
}
func (this *BaseInstaller) readPublicAccessLogPolicyPath() (string, error) {
policyId, err := models.SharedHTTPAccessLogPolicyDAO.FindCurrentPublicPolicyId(nil)
if err != nil {
return "", fmt.Errorf("find current public access log policy failed: %w", err)
}
if policyId <= 0 {
return "", nil
}
policy, err := models.SharedHTTPAccessLogPolicyDAO.FindEnabledHTTPAccessLogPolicy(nil, policyId)
if err != nil {
return "", fmt.Errorf("read public access log policy failed: %w", err)
}
if policy == nil {
return "", nil
}
return strings.TrimSpace(models.ParseHTTPAccessLogPolicyFilePath(policy)), nil
}
func dirFromPolicyPath(policyPath string) string {
pathValue := strings.TrimSpace(policyPath)
if pathValue == "" {
return ""
}
pathValue = strings.ReplaceAll(pathValue, "\\", "/")
dir := slashpath.Dir(pathValue)
if dir == "." {
return ""
}
return strings.TrimRight(dir, "/")
}
func (this *BaseInstaller) applyManagedConfig(tempDir string, desired *fluentBitDesiredConfig, parserContent string, existingMeta *fluentBitManagedMeta) (bool, error) {
mainExists, err := this.remoteFileExists(fluentBitMainConfigFile)
if err != nil {
return false, err
}
if mainExists && existingMeta == nil {
containsMarker, err := this.remoteFileContains(fluentBitMainConfigFile, fluentBitManagedMarker)
if err != nil {
return false, err
}
if !containsMarker {
// Adopt unmanaged config by backing it up and replacing with managed config below.
}
}
configContent, err := renderManagedConfig(desired)
if err != nil {
return false, err
}
envContent := renderManagedEnv(desired.ClickHouse)
metaContent, newMeta, err := renderManagedMeta(desired, configContent, parserContent, envContent)
if err != nil {
return false, err
}
requiredFiles := []string{fluentBitMainConfigFile, fluentBitParsersFile, fluentBitManagedEnvFile, fluentBitManagedMetaFile}
if existingMeta != nil && existingMeta.Hash == newMeta.Hash {
allExists := true
for _, file := range requiredFiles {
exists, err := this.remoteFileExists(file)
if err != nil {
return false, err
}
if !exists {
allExists = false
break
}
}
if allExists {
return false, nil
}
}
if mainExists {
backup := fluentBitMainConfigFile + ".bak." + strconv.FormatInt(time.Now().Unix(), 10)
_, stderr, err := this.client.Exec("cp -f " + shQuote(fluentBitMainConfigFile) + " " + shQuote(backup))
if err != nil {
return false, fmt.Errorf("backup existing fluent-bit config failed: %w, stderr: %s", err, stderr)
}
}
if err := this.writeRemoteFileByTemp(tempDir, fluentBitMainConfigFile, configContent, 0644); err != nil {
return false, err
}
if err := this.writeRemoteFileByTemp(tempDir, fluentBitParsersFile, parserContent, 0644); err != nil {
return false, err
}
if err := this.writeRemoteFileByTemp(tempDir, fluentBitManagedEnvFile, envContent, 0600); err != nil {
return false, err
}
if err := this.writeRemoteFileByTemp(tempDir, fluentBitManagedMetaFile, metaContent, 0644); err != nil {
return false, err
}
return true, nil
}
func renderManagedConfig(desired *fluentBitDesiredConfig) (string, error) {
if desired == nil || desired.ClickHouse == nil {
return "", errors.New("invalid fluent-bit desired config")
}
scheme := strings.ToLower(strings.TrimSpace(desired.ClickHouse.Scheme))
if scheme == "" {
scheme = "http"
}
if scheme != "http" && scheme != "https" {
return "", fmt.Errorf("unsupported clickhouse scheme '%s'", desired.ClickHouse.Scheme)
}
useTLS := scheme == "https"
insertHTTP := url.QueryEscape(fmt.Sprintf("INSERT INTO %s.logs_ingest FORMAT JSONEachRow", desired.ClickHouse.Database))
insertDNS := url.QueryEscape(fmt.Sprintf("INSERT INTO %s.dns_logs_ingest FORMAT JSONEachRow", desired.ClickHouse.Database))
insertHTTPDNS := url.QueryEscape(fmt.Sprintf("INSERT INTO %s.httpdns_access_logs_ingest FORMAT JSONEachRow", desired.ClickHouse.Database))
lines := []string{
"# " + fluentBitManagedMarker,
"[SERVICE]",
" Flush 1",
" Log_Level info",
" Parsers_File " + fluentBitParsersFile,
" storage.path " + fluentBitStorageDir,
" storage.sync normal",
" storage.checksum off",
" storage.backlog.mem_limit 512MB",
"",
}
if hasRole(desired.Roles, fluentBitRoleNode) {
lines = append(lines,
"[INPUT]",
" Name tail",
" Path "+desired.HTTPPathPattern,
" Tag app.http.logs",
" Parser json",
" Refresh_Interval 2",
" Read_from_Head false",
" DB /var/lib/fluent-bit/http-logs.db",
" storage.type filesystem",
" Mem_Buf_Limit 256MB",
" Skip_Long_Lines On",
"",
)
}
if hasRole(desired.Roles, fluentBitRoleDNS) {
lines = append(lines,
"[INPUT]",
" Name tail",
" Path "+desired.DNSPathPattern,
" Tag app.dns.logs",
" Parser json",
" Refresh_Interval 2",
" Read_from_Head false",
" DB /var/lib/fluent-bit/dns-logs.db",
" storage.type filesystem",
" Mem_Buf_Limit 256MB",
" Skip_Long_Lines On",
"",
)
}
if hasRole(desired.Roles, fluentBitRoleHTTPDNS) {
lines = append(lines,
"[INPUT]",
" Name tail",
" Path "+desired.HTTPDNSPathPattern,
" Tag app.httpdns.logs",
" Parser json",
" Refresh_Interval 2",
" Read_from_Head false",
" DB /var/lib/fluent-bit/httpdns-logs.db",
" storage.type filesystem",
" Mem_Buf_Limit 256MB",
" Skip_Long_Lines On",
"",
)
}
if hasRole(desired.Roles, fluentBitRoleNode) {
lines = append(lines,
"[OUTPUT]",
" Name http",
" Match app.http.logs",
" Host "+desired.ClickHouse.Host,
" Port "+strconv.Itoa(desired.ClickHouse.Port),
" URI /?query="+insertHTTP,
" Format json_lines",
" http_user ${CH_USER}",
" http_passwd ${CH_PASSWORD}",
" json_date_key timestamp",
" json_date_format epoch",
" workers 2",
" net.keepalive On",
" Retry_Limit False",
)
if useTLS {
lines = append(lines, " tls On")
if desired.ClickHouse.TLSSkipVerify {
lines = append(lines, " tls.verify Off")
} else {
lines = append(lines, " tls.verify On")
}
if strings.TrimSpace(desired.ClickHouse.TLSServerName) != "" {
lines = append(lines, " tls.vhost "+strings.TrimSpace(desired.ClickHouse.TLSServerName))
}
}
lines = append(lines, "")
}
if hasRole(desired.Roles, fluentBitRoleDNS) {
lines = append(lines,
"[OUTPUT]",
" Name http",
" Match app.dns.logs",
" Host "+desired.ClickHouse.Host,
" Port "+strconv.Itoa(desired.ClickHouse.Port),
" URI /?query="+insertDNS,
" Format json_lines",
" http_user ${CH_USER}",
" http_passwd ${CH_PASSWORD}",
" json_date_key timestamp",
" json_date_format epoch",
" workers 2",
" net.keepalive On",
" Retry_Limit False",
)
if useTLS {
lines = append(lines, " tls On")
if desired.ClickHouse.TLSSkipVerify {
lines = append(lines, " tls.verify Off")
} else {
lines = append(lines, " tls.verify On")
}
if strings.TrimSpace(desired.ClickHouse.TLSServerName) != "" {
lines = append(lines, " tls.vhost "+strings.TrimSpace(desired.ClickHouse.TLSServerName))
}
}
lines = append(lines, "")
}
if hasRole(desired.Roles, fluentBitRoleHTTPDNS) {
lines = append(lines,
"[OUTPUT]",
" Name http",
" Match app.httpdns.logs",
" Host "+desired.ClickHouse.Host,
" Port "+strconv.Itoa(desired.ClickHouse.Port),
" URI /?query="+insertHTTPDNS,
" Format json_lines",
" http_user ${CH_USER}",
" http_passwd ${CH_PASSWORD}",
" json_date_key timestamp",
" json_date_format epoch",
" workers 2",
" net.keepalive On",
" Retry_Limit False",
)
if useTLS {
lines = append(lines, " tls On")
if desired.ClickHouse.TLSSkipVerify {
lines = append(lines, " tls.verify Off")
} else {
lines = append(lines, " tls.verify On")
}
if strings.TrimSpace(desired.ClickHouse.TLSServerName) != "" {
lines = append(lines, " tls.vhost "+strings.TrimSpace(desired.ClickHouse.TLSServerName))
}
}
lines = append(lines, "")
}
return strings.Join(lines, "\n"), nil
}
func renderManagedEnv(ch *systemconfigs.ClickHouseSetting) string {
user := "default"
password := ""
if ch != nil {
if strings.TrimSpace(ch.User) != "" {
user = strings.TrimSpace(ch.User)
}
password = ch.Password
}
return "CH_USER=" + strconv.Quote(user) + "\n" +
"CH_PASSWORD=" + strconv.Quote(password) + "\n"
}
func renderManagedMeta(desired *fluentBitDesiredConfig, configContent string, parserContent string, envContent string) (string, *fluentBitManagedMeta, error) {
hashInput := configContent + "\n---\n" + parserContent + "\n---\n" + envContent + "\n---\n" + strings.Join(desired.Roles, ",")
hashBytes := sha256.Sum256([]byte(hashInput))
hash := fmt.Sprintf("%x", hashBytes[:])
meta := &fluentBitManagedMeta{
Roles: desired.Roles,
Hash: hash,
UpdatedAt: time.Now().Unix(),
SourceVersion: teaconst.Version,
}
data, err := json.MarshalIndent(meta, "", " ")
if err != nil {
return "", nil, fmt.Errorf("encode fluent-bit managed metadata failed: %w", err)
}
return string(data) + "\n", meta, nil
}
func (this *BaseInstaller) copyLocalFileToRemote(tempDir string, localPath string, remotePath string, mode os.FileMode) error {
tempFile := tempDir + "/" + filepath.Base(remotePath)
if err := this.client.Copy(localPath, tempFile, mode); err != nil {
return fmt.Errorf("upload fluent-bit file '%s' failed: %w", localPath, err)
}
_, stderr, err := this.client.Exec("cp -f " + shQuote(tempFile) + " " + shQuote(remotePath) + " && chmod " + fmt.Sprintf("%04o", mode) + " " + shQuote(remotePath))
if err != nil {
return fmt.Errorf("install remote fluent-bit file '%s' failed: %w, stderr: %s", remotePath, err, stderr)
}
return nil
}
func (this *BaseInstaller) writeRemoteFileByTemp(tempDir string, remotePath string, content string, mode os.FileMode) error {
tempFile := tempDir + "/" + filepath.Base(remotePath) + ".tmp"
if _, err := this.client.WriteFile(tempFile, []byte(content)); err != nil {
return fmt.Errorf("write temp fluent-bit file '%s' failed: %w", tempFile, err)
}
_, stderr, err := this.client.Exec("cp -f " + shQuote(tempFile) + " " + shQuote(remotePath) + " && chmod " + fmt.Sprintf("%04o", mode) + " " + shQuote(remotePath))
if err != nil {
return fmt.Errorf("write managed fluent-bit file '%s' failed: %w, stderr: %s", remotePath, err, stderr)
}
return nil
}
func (this *BaseInstaller) ensureFluentBitService(tempDir string, binPath string, configChanged bool) error {
_, _, _ = this.client.Exec("if command -v systemctl >/dev/null 2>&1 && [ ! -f /etc/systemd/system/" + fluentBitServiceName + ".service ] && [ ! -f /lib/systemd/system/" + fluentBitServiceName + ".service ]; then " +
"cat > /etc/systemd/system/" + fluentBitServiceName + ".service <<'EOF'\n" +
"[Unit]\n" +
"Description=Fluent Bit\n" +
"After=network.target\n" +
"\n" +
"[Service]\n" +
"ExecStart=" + binPath + " -c " + fluentBitMainConfigFile + "\n" +
"Restart=always\n" +
"RestartSec=5\n" +
"\n" +
"[Install]\n" +
"WantedBy=multi-user.target\n" +
"EOF\n" +
"fi")
stdout, _, err := this.client.Exec("if command -v systemctl >/dev/null 2>&1; then echo 1; else echo 0; fi")
if err != nil {
return fmt.Errorf("check systemctl failed: %w", err)
}
if strings.TrimSpace(stdout) == "1" {
dropInChanged, err := this.ensureServiceDropIn(tempDir, binPath)
if err != nil {
return err
}
restartRequired := configChanged || dropInChanged
_, stderr, err := this.client.Exec("systemctl daemon-reload; systemctl enable " + fluentBitServiceName + " >/dev/null 2>&1 || true; " +
"if systemctl is-active " + fluentBitServiceName + " >/dev/null 2>&1; then " +
"if [ \"" + boolToString(restartRequired) + "\" = \"1\" ]; then systemctl restart " + fluentBitServiceName + "; fi; " +
"else systemctl start " + fluentBitServiceName + "; fi")
if err != nil {
return fmt.Errorf("ensure fluent-bit service failed: %w, stderr: %s", err, stderr)
}
return nil
}
if configChanged {
_, _, _ = this.client.Exec("pkill -f \"fluent-bit.*fluent-bit.conf\" >/dev/null 2>&1 || true")
}
_, _, runningErr := this.client.Exec("pgrep -f \"fluent-bit.*fluent-bit.conf\" >/dev/null 2>&1")
if runningErr != nil {
startCmd := "set -a; [ -f " + shQuote(fluentBitManagedEnvFile) + " ] && . " + shQuote(fluentBitManagedEnvFile) + "; set +a; " +
shQuote(binPath) + " -c " + shQuote(fluentBitMainConfigFile) + " >/dev/null 2>&1 &"
_, stderr, err := this.client.Exec(startCmd)
if err != nil {
return fmt.Errorf("start fluent-bit without systemd failed: %w, stderr: %s", err, stderr)
}
}
return nil
}
func (this *BaseInstaller) ensureServiceDropIn(tempDir string, binPath string) (bool, error) {
_, stderr, err := this.client.Exec("mkdir -p " + shQuote(fluentBitDropInDir))
if err != nil {
return false, fmt.Errorf("prepare fluent-bit drop-in dir failed: %w, stderr: %s", err, stderr)
}
content := "[Service]\n" +
"EnvironmentFile=-" + fluentBitManagedEnvFile + "\n" +
"ExecStart=\n" +
"ExecStart=" + binPath + " -c " + fluentBitMainConfigFile + "\n"
existing, _, _ := this.client.Exec("if [ -f " + shQuote(fluentBitDropInFile) + " ]; then cat " + shQuote(fluentBitDropInFile) + "; fi")
if existing == content {
return false, nil
}
if err := this.writeRemoteFileByTemp(tempDir, fluentBitDropInFile, content, 0644); err != nil {
return false, err
}
return true, nil
}
func (this *BaseInstaller) remoteFileExists(path string) (bool, error) {
stdout, stderr, err := this.client.Exec("if [ -f " + shQuote(path) + " ]; then echo 1; else echo 0; fi")
if err != nil {
return false, fmt.Errorf("check remote file '%s' failed: %w, stderr: %s", path, err, stderr)
}
return strings.TrimSpace(stdout) == "1", nil
}
func (this *BaseInstaller) remoteFileContains(path string, pattern string) (bool, error) {
stdout, stderr, err := this.client.Exec("if grep -F " + shQuote(pattern) + " " + shQuote(path) + " >/dev/null 2>&1; then echo 1; else echo 0; fi")
if err != nil {
return false, fmt.Errorf("check remote file content '%s' failed: %w, stderr: %s", path, err, stderr)
}
return strings.TrimSpace(stdout) == "1", nil
}
func shQuote(value string) string {
if value == "" {
return "''"
}
return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'"
}
func boolToString(v bool) string {
if v {
return "1"
}
return "0"
}

24
EdgeAPI/internal/installers/installer_base.go
View File

@@ -279,3 +279,27 @@ func (this *BaseInstaller) uname() (uname string) {
return "x86_64 GNU/Linux"
}
func parseOSRelease(content string) map[string]string {
result := map[string]string{}
lines := strings.Split(content, "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" || strings.HasPrefix(line, "#") || !strings.Contains(line, "=") {
continue
}
parts := strings.SplitN(line, "=", 2)
key := strings.TrimSpace(parts[0])
value := strings.TrimSpace(parts[1])
value = strings.Trim(value, "\"")
result[key] = value
}
return result
}
func shQuote(s string) string {
if len(s) == 0 {
return "''"
}
return "'" + strings.ReplaceAll(s, "'", "'\"'\"'") + "'"
}

5
EdgeAPI/internal/installers/installer_httpdns_node.go
View File

@@ -131,11 +131,6 @@ https.key: "${keyFile}"`)
return fmt.Errorf("write '%s': %w", configFile, err)
}
err = i.SetupFluentBit(nodeconfigs.NodeRoleHTTPDNS)
if err != nil {
installStatus.ErrorCode = "SETUP_FLUENT_BIT_FAILED"
return fmt.Errorf("setup fluent-bit failed: %w", err)
}
startCmdPrefix := "cd " + shQuote(appDir+"/configs") + " && ../bin/edge-httpdns "

6
EdgeAPI/internal/installers/installer_node.go
View File

@@ -137,12 +137,6 @@ secret: "${nodeSecret}"`)
}
}
// 在线安装/更新 Fluent Bit（与边缘节点安装流程联动）
err = this.SetupFluentBit(nodeconfigs.NodeRoleNode)
if err != nil {
installStatus.ErrorCode = "SETUP_FLUENT_BIT_FAILED"
return fmt.Errorf("setup fluent-bit failed: %w", err)
}
// 测试
_, stderr, err = this.client.Exec(dir + "/edge-node/bin/edge-node test")

6
EdgeAPI/internal/installers/installer_ns_node_plus.go
View File

@@ -139,12 +139,6 @@ secret: "${nodeSecret}"`)
}
}
// 在线安装/更新 Fluent Bit（与 DNS 节点安装流程联动）
err = this.SetupFluentBit(nodeconfigs.NodeRoleDNS)
if err != nil {
installStatus.ErrorCode = "SETUP_FLUENT_BIT_FAILED"
return fmt.Errorf("setup fluent-bit failed: %w", err)
}
// 测试
_, stderr, err = this.client.Exec(dir + "/edge-dns/bin/edge-dns test")

6
EdgeAPI/internal/nodes/api_node_services_hook_plus.go
View File

@@ -7,6 +7,7 @@ import (
"github.com/TeaOSLab/EdgeAPI/internal/rpc/services"
"github.com/TeaOSLab/EdgeAPI/internal/rpc/services/accounts"
"github.com/TeaOSLab/EdgeAPI/internal/rpc/services/anti-ddos"
httpdnsServices "github.com/TeaOSLab/EdgeAPI/internal/rpc/services/httpdns"
"github.com/TeaOSLab/EdgeAPI/internal/rpc/services/nameservers"
"github.com/TeaOSLab/EdgeAPI/internal/rpc/services/posts"
"github.com/TeaOSLab/EdgeAPI/internal/rpc/services/reporters"
@@ -313,4 +314,9 @@ func APINodeServicesRegister(node *APINode, server *grpc.Server) {
pb.RegisterPostCategoryServiceServer(server, instance)
node.rest(instance)
}
{
var instance = node.serviceInstance(&httpdnsServices.HTTPDNSBoardService{}).(*httpdnsServices.HTTPDNSBoardService)
pb.RegisterHTTPDNSBoardServiceServer(server, instance)
node.rest(instance)
}
}

324
EdgeAPI/internal/rpc/services/httpdns/service_httpdns_board.go Normal file
View File

@@ -0,0 +1,324 @@
package httpdns
import (
"context"
"time"
"github.com/TeaOSLab/EdgeAPI/internal/clickhouse"
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"github.com/TeaOSLab/EdgeAPI/internal/rpc/services"
"github.com/TeaOSLab/EdgeAPI/internal/utils"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/iwind/TeaGo/dbs"
timeutil "github.com/iwind/TeaGo/utils/time"
)
// HTTPDNSBoardService HTTPDNS 仪表盘服务
type HTTPDNSBoardService struct {
services.BaseService
}
// ComposeHTTPDNSBoard 组合看板数据
func (this *HTTPDNSBoardService) ComposeHTTPDNSBoard(ctx context.Context, req *pb.ComposeHTTPDNSBoardRequest) (*pb.ComposeHTTPDNSBoardResponse, error) {
_, err := this.ValidateAdmin(ctx)
if err != nil {
return nil, err
}
tx := this.NullTx()
result := &pb.ComposeHTTPDNSBoardResponse{}
countApps, err := models.SharedHTTPDNSAppDAO.CountEnabledApps(tx, "")
if err != nil {
return nil, err
}
result.CountApps = countApps
countDomains, err := models.SharedHTTPDNSDomainDAO.CountEnabledDomains(tx, 0)
if err != nil {
return nil, err
}
result.CountDomains = countDomains
countClusters, err := models.SharedHTTPDNSClusterDAO.CountEnabledClusters(tx, "")
if err != nil {
return nil, err
}
result.CountClusters = countClusters
allNodes, err := models.SharedHTTPDNSNodeDAO.ListEnabledNodes(tx, 0)
if err != nil {
return nil, err
}
result.CountNodes = int64(len(allNodes))
var countOffline int64
for _, node := range allNodes {
if !node.IsActive {
countOffline++
}
}
result.CountOfflineNodes = countOffline
hourFrom := timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
hourTo := timeutil.Format("YmdH")
dayFrom := timeutil.Format("Ymd", time.Now().AddDate(0, 0, -14))
dayTo := timeutil.Format("Ymd")
todayFrom := timeutil.Format("Ymd", time.Now().Add(-24*time.Hour))
store := clickhouse.NewHTTPDNSAccessLogsStore()
if store.Client().IsConfigured() {
err = this.composeTrafficAndRanksFromClickHouse(ctx, tx, store, result, hourFrom, hourTo, dayFrom, dayTo, todayFrom)
}
if err != nil || !store.Client().IsConfigured() {
err = this.composeTrafficAndRanksFromMySQL(tx, result, hourFrom, hourTo, dayFrom, dayTo, todayFrom)
if err != nil {
return nil, err
}
}
err = this.fillNodeValues(tx, result)
if err != nil {
return nil, err
}
return result, nil
}
func (this *HTTPDNSBoardService) composeTrafficAndRanksFromClickHouse(ctx context.Context, tx *dbs.Tx, store *clickhouse.HTTPDNSAccessLogsStore, result *pb.ComposeHTTPDNSBoardResponse, hourFrom string, hourTo string, dayFrom string, dayTo string, todayFrom string) error {
hourlyStats, err := store.FindHourlyStats(ctx, hourFrom, hourTo)
if err != nil {
return err
}
hourlyMap := map[string]*clickhouse.HTTPDNSAccessLogHourlyStat{}
for _, stat := range hourlyStats {
hourlyMap[stat.Hour] = stat
}
hours, err := utils.RangeHours(hourFrom, hourTo)
if err != nil {
return err
}
for _, hour := range hours {
stat, ok := hourlyMap[hour]
if ok {
result.HourlyTrafficStats = append(result.HourlyTrafficStats, &pb.ComposeHTTPDNSBoardResponse_HourlyTrafficStat{
Hour: hour,
CountRequests: stat.CountRequests,
})
} else {
result.HourlyTrafficStats = append(result.HourlyTrafficStats, &pb.ComposeHTTPDNSBoardResponse_HourlyTrafficStat{Hour: hour})
}
}
dailyStats, err := store.FindDailyStats(ctx, dayFrom, dayTo)
if err != nil {
return err
}
dailyMap := map[string]*clickhouse.HTTPDNSAccessLogDailyStat{}
for _, stat := range dailyStats {
dailyMap[stat.Day] = stat
}
days, err := utils.RangeDays(dayFrom, dayTo)
if err != nil {
return err
}
for _, day := range days {
stat, ok := dailyMap[day]
if ok {
result.DailyTrafficStats = append(result.DailyTrafficStats, &pb.ComposeHTTPDNSBoardResponse_DailyTrafficStat{
Day: day,
CountRequests: stat.CountRequests,
})
} else {
result.DailyTrafficStats = append(result.DailyTrafficStats, &pb.ComposeHTTPDNSBoardResponse_DailyTrafficStat{Day: day})
}
}
topAppStats, err := store.ListTopApps(ctx, todayFrom, dayTo, 10)
if err != nil {
return err
}
for _, stat := range topAppStats {
appName := stat.AppName
if len(appName) == 0 {
appName = stat.AppId
}
result.TopAppStats = append(result.TopAppStats, &pb.ComposeHTTPDNSBoardResponse_TopAppStat{
AppId: 0,
AppName: appName,
CountRequests: stat.CountRequests,
})
}
topDomainStats, err := store.ListTopDomains(ctx, todayFrom, dayTo, 10)
if err != nil {
return err
}
for _, stat := range topDomainStats {
result.TopDomainStats = append(result.TopDomainStats, &pb.ComposeHTTPDNSBoardResponse_TopDomainStat{
DomainName: stat.Domain,
CountRequests: stat.CountRequests,
})
}
topNodeStats, err := store.ListTopNodes(ctx, todayFrom, dayTo, 10)
if err != nil {
return err
}
for _, stat := range topNodeStats {
nodeName := ""
node, nodeErr := models.SharedHTTPDNSNodeDAO.FindEnabledNode(tx, int64(stat.NodeId))
if nodeErr == nil && node != nil {
nodeName = node.Name
}
if len(nodeName) == 0 {
continue
}
result.TopNodeStats = append(result.TopNodeStats, &pb.ComposeHTTPDNSBoardResponse_TopNodeStat{
ClusterId: int64(stat.ClusterId),
NodeId: int64(stat.NodeId),
NodeName: nodeName,
CountRequests: stat.CountRequests,
})
}
return nil
}
func (this *HTTPDNSBoardService) composeTrafficAndRanksFromMySQL(tx *dbs.Tx, result *pb.ComposeHTTPDNSBoardResponse, hourFrom string, hourTo string, dayFrom string, dayTo string, todayFrom string) error {
hourlyStats, err := models.SharedHTTPDNSAccessLogDAO.FindHourlyStats(tx, hourFrom, hourTo)
if err != nil {
return err
}
hourlyMap := map[string]*models.HTTPDNSAccessLogHourlyStat{}
for _, stat := range hourlyStats {
hourlyMap[stat.Hour] = stat
}
hours, err := utils.RangeHours(hourFrom, hourTo)
if err != nil {
return err
}
for _, hour := range hours {
stat, ok := hourlyMap[hour]
if ok {
result.HourlyTrafficStats = append(result.HourlyTrafficStats, &pb.ComposeHTTPDNSBoardResponse_HourlyTrafficStat{
Hour: hour,
CountRequests: stat.CountRequests,
})
} else {
result.HourlyTrafficStats = append(result.HourlyTrafficStats, &pb.ComposeHTTPDNSBoardResponse_HourlyTrafficStat{Hour: hour})
}
}
dailyStats, err := models.SharedHTTPDNSAccessLogDAO.FindDailyStats(tx, dayFrom, dayTo)
if err != nil {
return err
}
dailyMap := map[string]*models.HTTPDNSAccessLogDailyStat{}
for _, stat := range dailyStats {
dailyMap[stat.Day] = stat
}
days, err := utils.RangeDays(dayFrom, dayTo)
if err != nil {
return err
}
for _, day := range days {
stat, ok := dailyMap[day]
if ok {
result.DailyTrafficStats = append(result.DailyTrafficStats, &pb.ComposeHTTPDNSBoardResponse_DailyTrafficStat{
Day: day,
CountRequests: stat.CountRequests,
})
} else {
result.DailyTrafficStats = append(result.DailyTrafficStats, &pb.ComposeHTTPDNSBoardResponse_DailyTrafficStat{Day: day})
}
}
topAppStats, err := models.SharedHTTPDNSAccessLogDAO.ListTopApps(tx, todayFrom, dayTo, 10)
if err != nil {
return err
}
for _, stat := range topAppStats {
appName := stat.AppName
if len(appName) == 0 {
appName = stat.AppId
}
result.TopAppStats = append(result.TopAppStats, &pb.ComposeHTTPDNSBoardResponse_TopAppStat{
AppId: 0,
AppName: appName,
CountRequests: stat.CountRequests,
})
}
topDomainStats, err := models.SharedHTTPDNSAccessLogDAO.ListTopDomains(tx, todayFrom, dayTo, 10)
if err != nil {
return err
}
for _, stat := range topDomainStats {
result.TopDomainStats = append(result.TopDomainStats, &pb.ComposeHTTPDNSBoardResponse_TopDomainStat{
DomainName: stat.Domain,
CountRequests: stat.CountRequests,
})
}
topNodeStats, err := models.SharedHTTPDNSAccessLogDAO.ListTopNodes(tx, todayFrom, dayTo, 10)
if err != nil {
return err
}
for _, stat := range topNodeStats {
nodeName := ""
node, nodeErr := models.SharedHTTPDNSNodeDAO.FindEnabledNode(tx, int64(stat.NodeId))
if nodeErr == nil && node != nil {
nodeName = node.Name
}
if len(nodeName) == 0 {
continue
}
result.TopNodeStats = append(result.TopNodeStats, &pb.ComposeHTTPDNSBoardResponse_TopNodeStat{
ClusterId: int64(stat.ClusterId),
NodeId: int64(stat.NodeId),
NodeName: nodeName,
CountRequests: stat.CountRequests,
})
}
return nil
}
func (this *HTTPDNSBoardService) fillNodeValues(tx *dbs.Tx, result *pb.ComposeHTTPDNSBoardResponse) error {
cpuValues, err := models.SharedNodeValueDAO.ListValuesForHTTPDNSNodes(tx, nodeconfigs.NodeValueItemCPU, "usage", nodeconfigs.NodeValueRangeMinute)
if err != nil {
return err
}
for _, v := range cpuValues {
result.CpuNodeValues = append(result.CpuNodeValues, &pb.NodeValue{
ValueJSON: v.Value,
CreatedAt: int64(v.CreatedAt),
})
}
memoryValues, err := models.SharedNodeValueDAO.ListValuesForHTTPDNSNodes(tx, nodeconfigs.NodeValueItemMemory, "usage", nodeconfigs.NodeValueRangeMinute)
if err != nil {
return err
}
for _, v := range memoryValues {
result.MemoryNodeValues = append(result.MemoryNodeValues, &pb.NodeValue{
ValueJSON: v.Value,
CreatedAt: int64(v.CreatedAt),
})
}
loadValues, err := models.SharedNodeValueDAO.ListValuesForHTTPDNSNodes(tx, nodeconfigs.NodeValueItemLoad, "load1m", nodeconfigs.NodeValueRangeMinute)
if err != nil {
return err
}
for _, v := range loadValues {
result.LoadNodeValues = append(result.LoadNodeValues, &pb.NodeValue{
ValueJSON: v.Value,
CreatedAt: int64(v.CreatedAt),
})
}
return nil
}

4
EdgeAPI/internal/rpc/services/service_node_value.go
View File

@@ -17,7 +17,7 @@ type NodeValueService struct {
// CreateNodeValue 记录数据
func (this *NodeValueService) CreateNodeValue(ctx context.Context, req *pb.CreateNodeValueRequest) (*pb.RPCSuccess, error) {
role, nodeId, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeNode, rpcutils.UserTypeDNS, rpcutils.UserTypeUser)
role, nodeId, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeNode, rpcutils.UserTypeDNS, rpcutils.UserTypeUser, rpcutils.UserTypeHTTPDNS)
if err != nil {
return nil, err
}
@@ -30,6 +30,8 @@ func (this *NodeValueService) CreateNodeValue(ctx context.Context, req *pb.Creat
clusterId, err = models.SharedNodeDAO.FindNodeClusterId(tx, nodeId)
case rpcutils.UserTypeDNS:
clusterId, err = models.SharedNSNodeDAO.FindNodeClusterId(tx, nodeId)
case rpcutils.UserTypeHTTPDNS:
clusterId, err = models.SharedHTTPDNSNodeDAO.FindNodeClusterId(tx, nodeId)
case rpcutils.UserTypeUser:
}
if err != nil {