=== 一、主机「测试连接」永远显示旧错误(真实 bug)===
HostService.Validate 的失败分支也写 status=active,且从不写 statusMessage:
if info, err := client.GetHostInfo(ctx); err == nil { 更新 os/arch }
else { s.Update(id, {status:"active"}) } // ← 失败也标记成功
结果:连不上也显示正常,页面永远停在几分钟前的旧错误文案,
用户改了 IP 再点测试,还以为"改了还是不行"。
改为区分成功/失败并写清原因:
成功 → status=active + os/arch + statusMessage="连接成功"
失败 → status=error + statusMessage=FormatSSHError(err)
=== 二、回环地址在平台里必然连不上 ===
平台跑在容器/Pod 内,127.0.0.1 指向平台自身。
- 新建主机时拦截:hostIP 为 127.0.0.1/localhost/::1 直接报错并解释原因
- 前端表单同步校验 + 文案
- SSH 错误映射补充:连接被拒绝时提示可能是地址填了回环
=== 三、主机模式采集系统日志(两处会导致采集失败)===
1. pipeline 一律套 cri 阶段:cri 只认容器运行时格式,
套在 /var/log/messages 这类 syslog 上会取不到时间戳、整行被丢。
→ appPipelineStages() 按路径判断:容器日志用 cri,普通文本日志只做限流。
2. promtailExternalLabels 写死 8 空格缩进:主机模板 external_labels 只有 4 空格,
一旦给采集器配置了标签就生成非法 YAML,Promtail 装不上。
→ 增加 indent 参数,主机 6 空格 / DaemonSet 8 空格。
3. 新增 system 采集任务:
/var/log/{messages,secure,cron,maillog,syslog,dmesg}
正则解析 syslog 时间戳(Jan _2 15:04:05,Asia/Shanghai)+ 限流
测试:TestHostPromtailConfigValid / CRIPathKeepsCRIStage / PlainFileHasNoCRIStage
TestPromtailManifestParses 全部通过
=== 四、资源用量 CPU 解析(nanocores)===
metrics-server 返回的 CPU 常是纳核(如 "194533645n"=194.5m)。
原解析只处理 "m",导致 194 亿核、页面出现"CPU 已达 limit 的 4863341125%"。
→ parseQuantityMilli 支持 n/u/m/整核
实测:CPU 当前=178m,峰值=3300m,建议"峰值占 limit 82%,可考虑提前扩容"
=== 五、其他 ===
- 添加集群默认改用 Token 认证(最小权限、可随时吊销)
- RBAC 片段补 metrics.k8s.io(资源面板需要,独立 API 组常被漏授)
- 文案去掉 Grafana 需求描述(平台自身不部署 Grafana)
镜像:lghost/lokistack:latest + v0.1 = sha256:7bddd0f4d1f1
229 lines
6.1 KiB
Go
229 lines
6.1 KiB
Go
package services
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"lokistack/models"
|
||
"lokistack/pkg/crypto"
|
||
"lokistack/repository"
|
||
"lokistack/utils"
|
||
)
|
||
|
||
type HostService struct{}
|
||
|
||
func NewHostService() *HostService {
|
||
return &HostService{}
|
||
}
|
||
|
||
// isLoopbackHost 判断是否回环地址。平台在容器/Pod 内运行,
|
||
// 回环地址指向平台自己,不可能连到目标服务器。
|
||
func isLoopbackHost(hostIP string) bool {
|
||
h := strings.ToLower(strings.TrimSpace(hostIP))
|
||
h = strings.TrimPrefix(h, "[")
|
||
h = strings.TrimSuffix(h, "]")
|
||
return h == "localhost" || h == "127.0.0.1" || h == "::1" ||
|
||
strings.HasPrefix(h, "127.")
|
||
}
|
||
|
||
func (s *HostService) Create(name, hostIP string, port int, user, authType, password, privateKey, keyPhrase, os, arch, labels string) (*models.Host, error) {
|
||
if port == 0 {
|
||
port = 22
|
||
}
|
||
// 平台自身跑在容器里,127.0.0.1 指向的是平台自己,永远连不上目标服务器。
|
||
// 这个坑很常见(本地 ssh root@127.0.0.1 能通,平台就报"连接被拒绝"),
|
||
// 所以在录入时就拦住,并说清原因。
|
||
if isLoopbackHost(hostIP) {
|
||
return nil, fmt.Errorf("主机地址不能填 %s:平台运行在容器内,127.0.0.1 指向平台自身。"+
|
||
"请填写目标服务器的实际 IP(或平台可达的域名)", hostIP)
|
||
}
|
||
// 密钥认证必须有私钥,否则存进去也连不上,只会在测试连接时报认证失败
|
||
if authType == "key" && privateKey == "" {
|
||
return nil, fmt.Errorf("认证方式选择了密钥,但没有提供私钥内容,请粘贴私钥后再保存")
|
||
}
|
||
if authType == "password" && password == "" {
|
||
return nil, fmt.Errorf("认证方式选择了密码,但没有填写密码")
|
||
}
|
||
host := models.Host{
|
||
ID: repository.GetDB().NextID("hosts"),
|
||
Name: name,
|
||
HostIP: hostIP,
|
||
Port: port,
|
||
User: user,
|
||
AuthType: authType,
|
||
OS: os,
|
||
Arch: arch,
|
||
Labels: labels,
|
||
Status: "pending",
|
||
CreatedAt: time.Now(),
|
||
UpdatedAt: time.Now(),
|
||
}
|
||
if password != "" {
|
||
enc, err := crypto.EncryptField(password)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("密码加密失败: %w", err)
|
||
}
|
||
host.Password = enc
|
||
}
|
||
if privateKey != "" {
|
||
enc, err := crypto.EncryptField(privateKey)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("私钥加密失败: %w", err)
|
||
}
|
||
host.PrivateKey = enc
|
||
}
|
||
if keyPhrase != "" {
|
||
enc, err := crypto.EncryptField(keyPhrase)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("私钥口令加密失败: %w", err)
|
||
}
|
||
host.KeyPhrase = enc
|
||
}
|
||
|
||
db := repository.GetDB()
|
||
db.Lock()
|
||
db.Hosts = append(db.Hosts, host)
|
||
db.Unlock()
|
||
db.RequestSave()
|
||
|
||
go s.Validate(host.ID)
|
||
return &host, nil
|
||
}
|
||
|
||
func (s *HostService) Get(id uint) (*models.Host, error) {
|
||
db := repository.GetDB()
|
||
db.RLock()
|
||
defer db.RUnlock()
|
||
for i := range db.Hosts {
|
||
if db.Hosts[i].ID == id {
|
||
h := db.Hosts[i]
|
||
return &h, nil
|
||
}
|
||
}
|
||
return nil, utils.ErrNotFound
|
||
}
|
||
|
||
func (s *HostService) List(page, pageSize int) ([]models.Host, int64, error) {
|
||
db := repository.GetDB()
|
||
db.RLock()
|
||
defer db.RUnlock()
|
||
|
||
total := int64(len(db.Hosts))
|
||
start := (page - 1) * pageSize
|
||
if start < 0 || start >= len(db.Hosts) {
|
||
return []models.Host{}, total, nil
|
||
}
|
||
end := start + pageSize
|
||
if end > len(db.Hosts) {
|
||
end = len(db.Hosts)
|
||
}
|
||
result := make([]models.Host, end-start)
|
||
copy(result, db.Hosts[start:end])
|
||
return result, total, nil
|
||
}
|
||
|
||
func (s *HostService) Update(id uint, updates map[string]interface{}) error {
|
||
db := repository.GetDB()
|
||
db.Lock()
|
||
defer db.Unlock()
|
||
for i := range db.Hosts {
|
||
if db.Hosts[i].ID == id {
|
||
if v, ok := updates["name"].(string); ok {
|
||
db.Hosts[i].Name = v
|
||
}
|
||
if v, ok := updates["status"].(string); ok {
|
||
db.Hosts[i].Status = v
|
||
}
|
||
if v, ok := updates["statusMessage"].(string); ok {
|
||
db.Hosts[i].StatusMessage = v
|
||
}
|
||
if v, ok := updates["os"].(string); ok {
|
||
db.Hosts[i].OS = v
|
||
}
|
||
if v, ok := updates["arch"].(string); ok {
|
||
db.Hosts[i].Arch = v
|
||
}
|
||
if v, ok := updates["hostIP"].(string); ok && v != "" {
|
||
db.Hosts[i].HostIP = v
|
||
}
|
||
if v, ok := updates["port"].(float64); ok && v > 0 {
|
||
db.Hosts[i].Port = int(v)
|
||
}
|
||
if v, ok := updates["user"].(string); ok && v != "" {
|
||
db.Hosts[i].User = v
|
||
}
|
||
if v, ok := updates["labels"].(string); ok {
|
||
db.Hosts[i].Labels = v
|
||
}
|
||
// 凭证编辑约定:非空才更新(留空=保留旧密文)
|
||
for _, field := range []string{"password", "privateKey", "keyPhrase"} {
|
||
if v, ok := updates[field].(string); ok && v != "" {
|
||
if enc, err := crypto.EncryptField(v); err == nil {
|
||
switch field {
|
||
case "password":
|
||
db.Hosts[i].Password = enc
|
||
case "privateKey":
|
||
db.Hosts[i].PrivateKey = enc
|
||
case "keyPhrase":
|
||
db.Hosts[i].KeyPhrase = enc
|
||
}
|
||
}
|
||
}
|
||
}
|
||
db.Hosts[i].UpdatedAt = time.Now()
|
||
db.RequestSave()
|
||
return nil
|
||
}
|
||
}
|
||
return utils.ErrNotFound
|
||
}
|
||
|
||
func (s *HostService) Delete(id uint) error {
|
||
db := repository.GetDB()
|
||
db.Lock()
|
||
defer db.Unlock()
|
||
for i, h := range db.Hosts {
|
||
if h.ID == id {
|
||
db.Hosts = append(db.Hosts[:i], db.Hosts[i+1:]...)
|
||
db.RequestSave()
|
||
return nil
|
||
}
|
||
}
|
||
return utils.ErrNotFound
|
||
}
|
||
|
||
func (s *HostService) Validate(id uint) {
|
||
host, err := s.Get(id)
|
||
if err != nil {
|
||
return
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||
defer cancel()
|
||
|
||
client, err := NewSSHClientWithAuth(HostSSHAuth(host))
|
||
if err != nil {
|
||
s.Update(id, map[string]interface{}{"status": "error", "statusMessage": err.Error()})
|
||
return
|
||
}
|
||
defer client.Close()
|
||
|
||
// 注意:这里原来失败分支也写 status=active 且不写 statusMessage,
|
||
// 结果是「连不上也显示为正常,页面永远停在旧错误文案」,
|
||
// 用户会以为改完地址仍然不行。必须区分成功/失败并写清原因。
|
||
if info, err := client.GetHostInfo(ctx); err == nil {
|
||
s.Update(id, map[string]interface{}{
|
||
"status": "active",
|
||
"os": info.OS,
|
||
"arch": info.Arch,
|
||
"statusMessage": "连接成功",
|
||
})
|
||
} else {
|
||
s.Update(id, map[string]interface{}{
|
||
"status": "error",
|
||
"statusMessage": FormatSSHError(err),
|
||
})
|
||
}
|
||
}
|