wangdefa/oci-portal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 18 Wiki Activity

发布 0.3.0：恢复 Chat 接口并增强云端事件通知
CI / test (push) Successful in 31s
Details
Release / release (push) Successful in 48s
Details

Browse Source
This commit is contained in:
wangdefa
2026-07-13 10:13:44 +08:00
parent 489cb49cb3
commit c7cc5616ed
31 changed files with 1682 additions and 1625 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/service/airesponses.go
+12 -4
View File
@@ -73,12 +73,20 @@ func RespPassthroughUsage(payload []byte) *aiwire.Usage {
var root struct {
Usage *aiwire.RespUsage `json:"usage"`
}
if json.Unmarshal(payload, &root) != nil || root.Usage == nil {
if json.Unmarshal(payload, &root) != nil {
return nil
}
usage := &aiwire.Usage{PromptTokens: root.Usage.InputTokens,
CompletionTokens: root.Usage.OutputTokens, TotalTokens: root.Usage.TotalTokens}
if cached := root.Usage.InputTokensDetails.CachedTokens; cached > 0 {
return usageFromResp(root.Usage)
}
// usageFromResp 换算 Responses usage 为 OpenAI 口径(缓存命中透传);nil 原样返回。
func usageFromResp(u *aiwire.RespUsage) *aiwire.Usage {
if u == nil {
return nil
}
usage := &aiwire.Usage{PromptTokens: u.InputTokens,
CompletionTokens: u.OutputTokens, TotalTokens: u.TotalTokens}
if cached := u.InputTokensDetails.CachedTokens; cached > 0 {
usage.PromptTokensDetails = &aiwire.PromptTokensDetails{CachedTokens: cached}
}
return usage
internal/service/alertrule.go
-305
View File
@@ -1,305 +0,0 @@
package service
import (
"context"
"errors"
"fmt"
"log"
"net/netip"
"slices"
"strings"
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"oci-portal/internal/model"
)
// 告警规则约束与命中记录保留期(窗口计数之外多留几天便于排查)。
const (
alertMaxThreshold = 100
alertMaxWindowMin = 1440
alertHitRetention = 7 * 24 * time.Hour
alertSourceIPIn = "in"
alertSourceIPNotIn = "notin"
)
// ErrInvalidAlertRule 标记规则字段非法,api 层映射 400。
var ErrInvalidAlertRule = fmt.Errorf("告警规则字段非法")
// ListAlertRules 返回全部告警规则(创建顺序)。
func (s *LogEventService) ListAlertRules(ctx context.Context) ([]model.AlertRule, error) {
var rules []model.AlertRule
if err := s.db.WithContext(ctx).Order("id").Find(&rules).Error; err != nil {
return nil, fmt.Errorf("list alert rules: %w", err)
}
return rules, nil
}
// CreateAlertRule 校验并创建规则。
func (s *LogEventService) CreateAlertRule(ctx context.Context, rule model.AlertRule) (model.AlertRule, error) {
if err := validateAlertRule(&rule); err != nil {
return model.AlertRule{}, err
}
rule.ID = 0
if err := s.db.WithContext(ctx).Create(&rule).Error; err != nil {
return model.AlertRule{}, fmt.Errorf("create alert rule: %w", err)
}
return rule, nil
}
// UpdateAlertRule 校验并整体覆盖规则(含启停)。
func (s *LogEventService) UpdateAlertRule(ctx context.Context, id uint, rule model.AlertRule) (model.AlertRule, error) {
if err := validateAlertRule(&rule); err != nil {
return model.AlertRule{}, err
}
var cur model.AlertRule
if err := s.db.WithContext(ctx).First(&cur, id).Error; err != nil {
return model.AlertRule{}, fmt.Errorf("find alert rule %d: %w", id, err)
}
rule.ID, rule.CreatedAt = cur.ID, cur.CreatedAt
if err := s.db.WithContext(ctx).Save(&rule).Error; err != nil {
return model.AlertRule{}, fmt.Errorf("update alert rule: %w", err)
}
return rule, nil
}
// DeleteAlertRule 删除规则及其命中记录。
func (s *LogEventService) DeleteAlertRule(ctx context.Context, id uint) error {
if err := s.db.WithContext(ctx).Delete(&model.AlertRule{}, id).Error; err != nil {
return fmt.Errorf("delete alert rule: %w", err)
}
if err := s.db.WithContext(ctx).Where("rule_id = ?", id).Delete(&model.AlertRuleHit{}).Error; err != nil {
return fmt.Errorf("delete alert rule hits: %w", err)
}
return nil
}
// validateAlertRule 校验字段并归一化;非法时返回含具体原因的 ErrInvalidAlertRule 包装。
func validateAlertRule(rule *model.AlertRule) error {
rule.Name = strings.TrimSpace(rule.Name)
if rule.Name == "" {
return fmt.Errorf("%w: 名称必填", ErrInvalidAlertRule)
}
if rule.SourceIPMode == "" {
rule.SourceIPMode = alertSourceIPIn
}
if rule.SourceIPMode != alertSourceIPIn && rule.SourceIPMode != alertSourceIPNotIn {
return fmt.Errorf("%w: 来源 IP 模式须为 in/notin", ErrInvalidAlertRule)
}
if rule.Threshold < 1 || rule.Threshold > alertMaxThreshold {
return fmt.Errorf("%w: 阈值须在 1-%d 之间", ErrInvalidAlertRule, alertMaxThreshold)
}
if rule.Threshold > 1 && (rule.WindowMinutes < 1 || rule.WindowMinutes > alertMaxWindowMin) {
return fmt.Errorf("%w: 阈值>1 时窗口须在 1-%d 分钟之间", ErrInvalidAlertRule, alertMaxWindowMin)
}
if rule.EventTypes != "" {
rule.EventTypes = normalizeCSV(rule.EventTypes)
}
return validateAlertRuleIPs(rule)
}
// validateAlertRuleIPs 归一化并校验来源 IP 列表(裸 IP 或 CIDR)。
func validateAlertRuleIPs(rule *model.AlertRule) error {
if rule.SourceIPs == "" {
return nil
}
rule.SourceIPs = normalizeCSV(rule.SourceIPs)
for _, item := range strings.Split(rule.SourceIPs, ",") {
if _, err := parseIPMatcher(item); err != nil {
return fmt.Errorf("%w: 来源 IP %q 不是合法的 IP 或 CIDR", ErrInvalidAlertRule, item)
}
}
return nil
}
// normalizeCSV 去除各项空白与空项后重组逗号分隔串。
func normalizeCSV(s string) string {
parts := strings.Split(s, ",")
out := parts[:0]
for _, p := range parts {
if p = strings.TrimSpace(p); p != "" {
out = append(out, p)
}
}
return strings.Join(out, ",")
}
// parseIPMatcher 把裸 IP 或 CIDR 解析为前缀(裸 IP 视为单地址前缀)。
func parseIPMatcher(item string) (netip.Prefix, error) {
if strings.Contains(item, "/") {
return netip.ParsePrefix(item)
}
addr, err := netip.ParseAddr(item)
if err != nil {
return netip.Prefix{}, err
}
return netip.PrefixFrom(addr, addr.BitLen()), nil
}
// ipListMatch 报告 ip 是否命中列表中的任一前缀;ip 解析失败视为未命中。
func ipListMatch(list, ip string) bool {
addr, err := netip.ParseAddr(ip)
if err != nil {
return false
}
for _, item := range strings.Split(list, ",") {
if p, err := parseIPMatcher(item); err == nil && p.Contains(addr) {
return true
}
}
return false
}
// ruleHits 报告事件是否命中规则的全部条件(AND 语义,空条件视为任意)。
func ruleHits(rule model.AlertRule, e *model.LogEvent, p parsedEvent) bool {
if rule.OciConfigID != 0 && rule.OciConfigID != e.OciConfigID {
return false
}
name := relayEventShortName(p.EventType)
if rule.EventTypes != "" && !slices.Contains(strings.Split(rule.EventTypes, ","), name) {
return false
}
if rule.ResourceMatch != "" && !strings.Contains(p.ResourceName, rule.ResourceMatch) {
return false
}
return ruleIPHits(rule, p.SourceIP)
}
// ruleIPHits 按模式判定来源 IP 条件:in 命中列表告警;notin 不在列表才告警,
// 事件缺 IP 字段时 notin 不告警(避免解析缺字段导致白名单误报)。
func ruleIPHits(rule model.AlertRule, ip string) bool {
if rule.SourceIPs == "" {
return true
}
if rule.SourceIPMode == alertSourceIPNotIn {
return ip != "" && !ipListMatch(rule.SourceIPs, ip)
}
return ipListMatch(rule.SourceIPs, ip)
}
// matchAlertRules 对一条已解析事件执行全部启用规则;任何内部错误只记日志,不影响解析主流程。
func (s *LogEventService) matchAlertRules(ctx context.Context, rules []model.AlertRule, e *model.LogEvent, p parsedEvent) {
if s.notifier == nil {
return
}
for _, rule := range rules {
if !rule.Enabled || !ruleHits(rule, e, p) {
continue
}
count, ok := s.recordAlertHit(ctx, rule, e)
if !ok || count < rule.Threshold || !s.alertCooldownPass(rule) {
continue
}
s.notifier.SendTemplateAsync("audit_alert", map[string]string{
"rule": rule.Name, "tenant": s.configAlias(ctx, e.OciConfigID),
"event": relayEventShortName(p.EventType), "resource": p.ResourceName,
"ip": p.SourceIP, "count": fmt.Sprint(count),
})
}
}
// recordAlertHit 落一条命中并返回窗口内累计次数;阈值 1 的规则免计数直接触发。
func (s *LogEventService) recordAlertHit(ctx context.Context, rule model.AlertRule, e *model.LogEvent) (int, bool) {
count, err := s.recordAlertHitTx(ctx, rule, e)
if err != nil {
if !errors.Is(err, gorm.ErrRecordNotFound) {
log.Printf("alert rule hit record: %v", err)
}
return 0, false
}
return count, true
}
func (s *LogEventService) recordAlertHitTx(ctx context.Context, rule model.AlertRule, event *model.LogEvent) (int, error) {
count := 0
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var err error
count, err = insertAlertHit(tx, rule, event)
return err
})
return count, err
}
// insertAlertHit 按 rule→event 锁顺序确认引用存在后插入并统计窗口命中。
func insertAlertHit(tx *gorm.DB, rule model.AlertRule, event *model.LogEvent) (int, error) {
if err := lockAlertRefs(tx, rule.ID, event.ID); err != nil {
return 0, err
}
if rule.Threshold <= 1 {
return 1, nil
}
now := time.Now()
hit := model.AlertRuleHit{RuleID: rule.ID, LogEventID: event.ID, HitAt: now}
if err := tx.Create(&hit).Error; err != nil {
return 0, fmt.Errorf("create alert rule hit: %w", err)
}
var count int64
cutoff := now.Add(-time.Duration(rule.WindowMinutes) * time.Minute)
err := tx.Model(&model.AlertRuleHit{}).
Where("rule_id = ? AND hit_at >= ?", rule.ID, cutoff).Count(&count).Error
if err != nil {
return 0, fmt.Errorf("count alert rule hits: %w", err)
}
return int(count), nil
}
func lockAlertRefs(tx *gorm.DB, ruleID, eventID uint) error {
var rule model.AlertRule
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").First(&rule, ruleID).Error; err != nil {
return fmt.Errorf("lock alert rule %d: %w", ruleID, err)
}
var event model.LogEvent
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").First(&event, eventID).Error; err != nil {
return fmt.Errorf("lock log event %d: %w", eventID, err)
}
return nil
}
// alertCooldownPass 报告规则是否已过冷却窗口;通过即记录本次发送时刻。
// 阈值 1 的规则无冷却(每次命中即时告警,与既有云端事件通知一致)。
func (s *LogEventService) alertCooldownPass(rule model.AlertRule) bool {
if rule.Threshold <= 1 {
return true
}
s.alertMu.Lock()
defer s.alertMu.Unlock()
window := time.Duration(rule.WindowMinutes) * time.Minute
if last, ok := s.alertSentAt[rule.ID]; ok && time.Since(last) < window {
return false
}
if s.alertSentAt == nil {
s.alertSentAt = map[uint]time.Time{}
}
s.alertSentAt[rule.ID] = time.Now()
return true
}
// ClearAlertCooldown 清除已删除租户规则的进程内冷却状态。
func (s *LogEventService) ClearAlertCooldown(ruleIDs []uint) {
s.alertMu.Lock()
defer s.alertMu.Unlock()
for _, id := range ruleIDs {
delete(s.alertSentAt, id)
}
}
// loadEnabledAlertRules 载入启用中的规则;失败时返回空集并记日志(解析主流程照常)。
func (s *LogEventService) loadEnabledAlertRules(ctx context.Context) []model.AlertRule {
var rules []model.AlertRule
err := s.db.WithContext(ctx).Where("enabled = ?", true).Order("id").Find(&rules).Error
if err != nil {
log.Printf("load alert rules: %v", err)
return nil
}
return rules
}
// cleanupAlertHits 删除保留期外的命中记录(随 cleanupOnce 周期执行)。
func (s *LogEventService) cleanupAlertHits(ctx context.Context) {
cutoff := time.Now().Add(-alertHitRetention)
if err := s.db.WithContext(ctx).Where("hit_at < ?", cutoff).Delete(&model.AlertRuleHit{}).Error; err != nil {
log.Printf("cleanup alert hits: %v", err)
}
}
internal/service/alertrule_test.go
-345
View File
@@ -1,345 +0,0 @@
package service
import (
"context"
"fmt"
"strings"
"testing"
"time"
"oci-portal/internal/crypto"
"oci-portal/internal/model"
)
func TestValidateAlertRule(t *testing.T) {
tests := []struct {
name string
rule model.AlertRule
wantErr string
check func(t *testing.T, r model.AlertRule)
}{
{name: "名称必填", rule: model.AlertRule{Threshold: 1}, wantErr: "名称"},
{name: "模式非法", rule: model.AlertRule{Name: "r", Threshold: 1, SourceIPMode: "any"}, wantErr: "in/notin"},
{name: "阈值越界", rule: model.AlertRule{Name: "r", Threshold: 101}, wantErr: "阈值"},
{name: "阈值>1须带窗口", rule: model.AlertRule{Name: "r", Threshold: 3}, wantErr: "窗口"},
{name: "IP 非法", rule: model.AlertRule{Name: "r", Threshold: 1, SourceIPs: "300.1.1.1"}, wantErr: "IP"},
{name: "CIDR 合法", rule: model.AlertRule{Name: "r", Threshold: 1, SourceIPs: "10.0.0.0/8, 1.2.3.4"},
check: func(t *testing.T, r model.AlertRule) {
if r.SourceIPs != "10.0.0.0/8,1.2.3.4" {
t.Errorf("SourceIPs = %q, 应去空白归一化", r.SourceIPs)
}
if r.SourceIPMode != alertSourceIPIn {
t.Errorf("SourceIPMode = %q, 应默认 in", r.SourceIPMode)
}
}},
{name: "事件清单归一化", rule: model.AlertRule{Name: "r", Threshold: 1, EventTypes: " TerminateInstance , CreateApiKey ,"},
check: func(t *testing.T, r model.AlertRule) {
if r.EventTypes != "TerminateInstance,CreateApiKey" {
t.Errorf("EventTypes = %q", r.EventTypes)
}
}},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
rule := tt.rule
err := validateAlertRule(&rule)
if tt.wantErr == "" {
if err != nil {
t.Fatalf("validateAlertRule: %v", err)
}
if tt.check != nil {
tt.check(t, rule)
}
return
}
if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
t.Fatalf("err = %v, want contains %q", err, tt.wantErr)
}
})
}
}
func TestRuleHits(t *testing.T) {
base := model.AlertRule{Name: "r", Threshold: 1, SourceIPMode: alertSourceIPIn}
ev := &model.LogEvent{OciConfigID: 7}
parsed := parsedEvent{
EventType: "com.oraclecloud.ComputeApi.TerminateInstance",
SourceIP: "203.0.113.8",
ResourceName: "web-server-1",
}
tests := []struct {
name string
mod func(r *model.AlertRule)
p *parsedEvent
want bool
}{
{name: "空条件全命中", mod: func(r *model.AlertRule) {}, want: true},
{name: "租户匹配", mod: func(r *model.AlertRule) { r.OciConfigID = 7 }, want: true},
{name: "租户不匹配", mod: func(r *model.AlertRule) { r.OciConfigID = 8 }, want: false},
{name: "事件短名命中", mod: func(r *model.AlertRule) { r.EventTypes = "LaunchInstance,TerminateInstance" }, want: true},
{name: "事件不在清单", mod: func(r *model.AlertRule) { r.EventTypes = "CreateUser" }, want: false},
{name: "资源子串命中", mod: func(r *model.AlertRule) { r.ResourceMatch = "web-" }, want: true},
{name: "资源不含", mod: func(r *model.AlertRule) { r.ResourceMatch = "db-" }, want: false},
{name: "IP in 命中 CIDR", mod: func(r *model.AlertRule) { r.SourceIPs = "203.0.113.0/24" }, want: true},
{name: "IP in 未命中", mod: func(r *model.AlertRule) { r.SourceIPs = "10.0.0.0/8" }, want: false},
{name: "IP notin 白名单外告警", mod: func(r *model.AlertRule) {
r.SourceIPs, r.SourceIPMode = "10.0.0.0/8", alertSourceIPNotIn
}, want: true},
{name: "IP notin 白名单内不告警", mod: func(r *model.AlertRule) {
r.SourceIPs, r.SourceIPMode = "203.0.113.8", alertSourceIPNotIn
}, want: false},
{name: "notin 事件缺 IP 不告警", mod: func(r *model.AlertRule) {
r.SourceIPs, r.SourceIPMode = "10.0.0.0/8", alertSourceIPNotIn
}, p: &parsedEvent{EventType: parsed.EventType}, want: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
rule := base
tt.mod(&rule)
p := parsed
if tt.p != nil {
p = *tt.p
}
if got := ruleHits(rule, ev, p); got != tt.want {
t.Errorf("ruleHits = %v, want %v", got, tt.want)
}
})
}
}
func TestAlertRuleCRUD(t *testing.T) {
svc, _, _ := newLogEventEnv(t)
ctx := context.Background()
created, err := svc.CreateAlertRule(ctx, model.AlertRule{Name: "非白名单终止", Enabled: true, Threshold: 1})
if err != nil {
t.Fatalf("create: %v", err)
}
if created.ID == 0 {
t.Fatal("create 未回填 ID")
}
if _, err := svc.CreateAlertRule(ctx, model.AlertRule{Threshold: 1}); err == nil {
t.Fatal("空名称应校验失败")
}
created.Enabled = false
created.EventTypes = "TerminateInstance"
updated, err := svc.UpdateAlertRule(ctx, created.ID, created)
if err != nil {
t.Fatalf("update: %v", err)
}
if updated.Enabled || updated.EventTypes != "TerminateInstance" {
t.Fatalf("update 未生效: %+v", updated)
}
rules, err := svc.ListAlertRules(ctx)
if err != nil || len(rules) != 1 {
t.Fatalf("list = %v, %v", rules, err)
}
if err := svc.DeleteAlertRule(ctx, created.ID); err != nil {
t.Fatalf("delete: %v", err)
}
if rules, _ := svc.ListAlertRules(ctx); len(rules) != 0 {
t.Fatalf("delete 后仍有 %d 条", len(rules))
}
}
// auditEventPayload 构造一条含资源与来源 IP 的 CloudEvents 审计消息。
func auditEventPayload(event, resource, ip string) string {
return fmt.Sprintf(`{"eventType":"com.oraclecloud.ComputeApi.%s","source":"ComputeApi",`+
`"eventTime":"2026-07-10T08:00:00Z","data":{"resourceName":%q,"identity":{"ipAddress":%q}}}`,
event, resource, ip)
}
// newAlertNotifyEnv 组装带假 Telegram 通道的告警测试环境。
func newAlertNotifyEnv(t *testing.T) (*LogEventService, *telegramCapture, func()) {
t.Helper()
svc, db, _ := newLogEventEnv(t)
srv, rec := newFakeTelegram(t, `{"ok":true}`)
cipher, err := crypto.NewCipher("test-data-key")
if err != nil {
t.Fatalf("new cipher: %v", err)
}
settings := NewSettingService(db, cipher)
token := "123456:AAfake"
if err := settings.UpdateTelegram(context.Background(),
UpdateTelegramInput{Enabled: true, BotToken: &token, ChatID: "42"}); err != nil {
t.Fatalf("update telegram: %v", err)
}
n := NewNotifier(settings)
n.base = srv.URL
svc.SetNotifier(n, settings)
return svc, rec, n.Wait
}
func TestMatchAlertRulesNotify(t *testing.T) {
svc, rec, wait := newAlertNotifyEnv(t)
ctx := context.Background()
_, err := svc.CreateAlertRule(ctx, model.AlertRule{
Name: "白名单外终止", Enabled: true, Threshold: 1,
EventTypes: "TerminateInstance", SourceIPs: "10.0.0.0/8", SourceIPMode: alertSourceIPNotIn,
})
if err != nil {
t.Fatalf("create rule: %v", err)
}
// 命中:白名单外 IP;不命中:白名单内 IP
mustIngest(t, svc, "m1", auditEventPayload("TerminateInstance", "web-1", "203.0.113.8"))
mustIngest(t, svc, "m2", auditEventPayload("TerminateInstance", "web-2", "10.1.2.3"))
svc.parseOnce(ctx)
wait()
alerts := auditAlerts(rec.snapshot())
joined := strings.Join(alerts, "\n---\n")
if !strings.Contains(joined, "白名单外终止") || !strings.Contains(joined, "web-1") {
t.Fatalf("应收到含规则名与资源的告警,got %q", joined)
}
if strings.Contains(joined, "web-2") {
t.Fatalf("白名单内事件不应告警,got %q", joined)
}
}
// auditAlerts 过滤出审计告警推送(排除既有 notifyCritical 的云端事件通知)。
func auditAlerts(texts []string) []string {
var out []string
for _, s := range texts {
if strings.Contains(s, "审计告警") {
out = append(out, s)
}
}
return out
}
func TestAlertThresholdWindow(t *testing.T) {
svc, rec, wait := newAlertNotifyEnv(t)
ctx := context.Background()
_, err := svc.CreateAlertRule(ctx, model.AlertRule{
Name: "登录风暴", Enabled: true, Threshold: 3, WindowMinutes: 5, EventTypes: "InteractiveLogin",
})
if err != nil {
t.Fatalf("create rule: %v", err)
}
for i := 1; i <= 4; i++ {
mustIngest(t, svc, fmt.Sprint("login-", i),
auditEventPayload("InteractiveLogin", "user@x.com", "203.0.113.8"))
}
svc.parseOnce(ctx)
wait()
alerts := auditAlerts(rec.snapshot())
if len(alerts) != 1 {
t.Fatalf("窗口内 4 次命中应只告警 1 次(第 3 次触发后冷却),got %d 条: %v", len(alerts), alerts)
}
if !strings.Contains(alerts[0], "3 次") {
t.Errorf("告警文案应含累计次数,got %q", alerts[0])
}
}
// TestAlertRuleBadDataDoesNotBlockParse 验证规则表异常不影响解析主流程。
func TestAlertRuleBadDataDoesNotBlockParse(t *testing.T) {
svc, db, _ := newLogEventEnv(t)
ctx := context.Background()
// 直插一条绕过校验的坏规则(IP 列表非法)
bad := model.AlertRule{Name: "bad", Enabled: true, Threshold: 1, SourceIPs: "not-an-ip"}
if err := db.Create(&bad).Error; err != nil {
t.Fatalf("insert bad rule: %v", err)
}
mustIngest(t, svc, "m1", auditEventPayload("TerminateInstance", "web-1", "1.2.3.4"))
svc.parseOnce(ctx)
var e model.LogEvent
if err := db.First(&e, "message_id = ?", "m1").Error; err != nil {
t.Fatalf("find event: %v", err)
}
if !e.Processed {
t.Fatal("坏规则不应阻塞事件解析")
}
}
// mustIngest 落一条回传事件,失败即终止测试。
func mustIngest(t *testing.T, svc *LogEventService, msgID, payload string) {
t.Helper()
if err := svc.Ingest(context.Background(), 1, msgID, []byte(payload), false); err != nil {
t.Fatalf("ingest %s: %v", msgID, err)
}
}
// TestCleanupAlertHits 验证过期命中记录随清理删除。
func TestCleanupAlertHits(t *testing.T) {
svc, db, _ := newLogEventEnv(t)
old := model.AlertRuleHit{RuleID: 1, HitAt: time.Now().Add(-8 * 24 * time.Hour)}
fresh := model.AlertRuleHit{RuleID: 1, HitAt: time.Now()}
if err := db.Create(&old).Error; err != nil {
t.Fatalf("insert: %v", err)
}
if err := db.Create(&fresh).Error; err != nil {
t.Fatalf("insert: %v", err)
}
svc.cleanupAlertHits(context.Background())
var count int64
db.Model(&model.AlertRuleHit{}).Count(&count)
if count != 1 {
t.Fatalf("清理后应剩 1 条,got %d", count)
}
}
func TestClearAlertCooldown(t *testing.T) {
svc := NewLogEventService(nil)
rule1 := model.AlertRule{ID: 1, Threshold: 2, WindowMinutes: 10}
rule2 := model.AlertRule{ID: 2, Threshold: 2, WindowMinutes: 10}
if !svc.alertCooldownPass(rule1) || !svc.alertCooldownPass(rule2) {
t.Fatal("首次命中应通过冷却检查")
}
svc.ClearAlertCooldown([]uint{rule1.ID})
if !svc.alertCooldownPass(rule1) {
t.Fatal("已清理规则应重新通过冷却检查")
}
if svc.alertCooldownPass(rule2) {
t.Fatal("未清理规则不应通过冷却检查")
}
}
func TestRecordAlertHitRejectsMissingRefs(t *testing.T) {
tests := []struct {
name string
deleteRule bool
}{
{name: "规则已删除", deleteRule: true},
{name: "事件已删除"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
testRecordAlertHitMissingRef(t, tt.deleteRule)
})
}
}
func testRecordAlertHitMissingRef(t *testing.T, deleteRule bool) {
t.Helper()
svc, db, cfgID := newLogEventEnv(t)
rule := model.AlertRule{Name: "r", OciConfigID: cfgID, Threshold: 2, WindowMinutes: 5}
event := model.LogEvent{OciConfigID: cfgID, MessageID: "m"}
if err := db.Create(&rule).Error; err != nil {
t.Fatalf("create rule: %v", err)
}
if err := db.Create(&event).Error; err != nil {
t.Fatalf("create event: %v", err)
}
var err error
if deleteRule {
err = db.Delete(&rule).Error
} else {
err = db.Delete(&event).Error
}
if err != nil {
t.Fatalf("delete reference: %v", err)
}
if count, ok := svc.recordAlertHit(context.Background(), rule, &event); ok || count != 0 {
t.Fatalf("record missing refs = (%d,%v), want (0,false)", count, ok)
}
var hits int64
db.Model(&model.AlertRuleHit{}).Count(&hits)
if hits != 0 {
t.Fatalf("orphan alert hits = %d, want 0", hits)
}
}
internal/service/anthresponses.go
+6 -7
View File
@@ -83,7 +83,7 @@ func anthInputItems(messages []aiwire.AnthMessage) ([]any, error) {
func anthBlockItem(role string, b aiwire.AnthBlock) (item, part map[string]any, err error) {
switch b.Type {
case "text":
return nil, anthTextPart(role, b.Text), nil
return nil, respTextPart(role, b.Text), nil
case "image":
part, err = anthImageInput(b.Source)
return nil, part, err
@@ -100,8 +100,9 @@ func anthBlockItem(role string, b aiwire.AnthBlock) (item, part map[string]any,
}
}
// anthTextPart 按角色选择 Responses 文本部件类型(assistant 历史为 output_text)。
func anthTextPart(role, text string) map[string]any {
// respTextPart 按角色选择 Responses 文本部件类型(assistant 历史为 output_text);
// Anthropic 与 Chat Completions 两条转换链共用。
func respTextPart(role, text string) map[string]any {
if role == "assistant" {
return map[string]any{"type": "output_text", "text": text}
}
@@ -219,8 +220,7 @@ func ResponsesToAnthropic(payload []byte, msgID string) (*aiwire.MessagesRespons
out.StopReason = "tool_use"
}
}
if resp.Status == "incomplete" && resp.IncompleteDetails != nil &&
resp.IncompleteDetails.Reason == "max_output_tokens" {
if respTruncated(&resp) {
out.StopReason = "max_tokens"
}
out.Usage = anthUsageFromResp(resp.Usage)
@@ -337,8 +337,7 @@ func (st *AnthRespBridge) finishFrom(resp *respPayload) {
return
}
st.usage = anthUsageFromResp(resp.Usage)
if resp.Status == "incomplete" && resp.IncompleteDetails != nil &&
resp.IncompleteDetails.Reason == "max_output_tokens" {
if respTruncated(resp) {
st.stopReason = "max_tokens"
}
}
internal/service/chatresponses.go
+360
View File
@@ -0,0 +1,360 @@
package service
import (
"encoding/json"
"fmt"
"strings"
"oci-portal/internal/aiwire"
)
// OpenAI Chat Completions ↔ OCI OpenAI 兼容面(/actions/v1/responses)直通转换。
// 端点定位 Tier 2 兼容层(承接只会说 CC 的存量客户端),机制与 anthresponses.go
// 同构。语义损失(README 披露):stop / seed / n / penalty 等无对应字段,忽略;
// 上游 reasoning 输出项与增量事件丢弃;非 function 工具类型拒绝。
// ChatToResponsesBody 把 Chat Completions 请求转为直通 body(强制 store:false)。
func ChatToResponsesBody(req aiwire.ChatRequest) ([]byte, error) {
input, instructions, err := chatInputItems(req.Messages)
if err != nil {
return nil, err
}
body := map[string]any{"model": req.Model, "input": input, "store": false}
if instructions != "" {
body["instructions"] = instructions
}
chatSampling(body, req)
if err := chatBodyTools(body, req); err != nil {
return nil, err
}
if tf := chatTextFormat(req.ResponseFormat); tf != nil {
body["text"] = map[string]any{"format": tf}
}
if req.ReasoningEffort != "" {
body["reasoning"] = map[string]string{"effort": strings.ToLower(req.ReasoningEffort)}
}
if req.Stream {
body["stream"] = true
}
return json.Marshal(body)
}
// chatSampling 装配采样与输出预算;max_completion_tokens 优先于已弃用的 max_tokens。
func chatSampling(body map[string]any, req aiwire.ChatRequest) {
if req.MaxCompletionTokens != nil {
body["max_output_tokens"] = *req.MaxCompletionTokens
} else if req.MaxTokens != nil {
body["max_output_tokens"] = *req.MaxTokens
}
if req.Temperature != nil {
body["temperature"] = *req.Temperature
}
if req.TopP != nil {
body["top_p"] = *req.TopP
}
if req.ParallelToolCalls != nil {
body["parallel_tool_calls"] = *req.ParallelToolCalls
}
}
// chatBodyTools 装配工具与 tool_choice;非 function 工具类型拒绝
// (web_search 等新能力仅在 Responses / Messages 端点供给)。
func chatBodyTools(body map[string]any, req aiwire.ChatRequest) error {
if len(req.Tools) == 0 {
return nil
}
tools := make([]map[string]any, 0, len(req.Tools))
for _, t := range req.Tools {
if t.Type != "function" {
return fmt.Errorf("不支持的工具类型 %q:该端点仅支持 function 工具", t.Type)
}
tools = append(tools, map[string]any{"type": "function", "name": t.Function.Name,
"description": t.Function.Description, "parameters": t.Function.Parameters})
}
body["tools"] = tools
if tc := chatRespToolChoice(req.ToolChoice); tc != nil {
body["tool_choice"] = tc
}
return nil
}
// chatInputItems 把消息序列展开为 Responses input 项与 instructions:
// system/developer 文本聚为 instructions;tool 消息为 function_call_output 项;
// 其余消息经 chatMessageItems 展开,保持相对顺序。
func chatInputItems(messages []aiwire.ChatMessage) ([]any, string, error) {
items := make([]any, 0, len(messages))
var sys []string
for _, m := range messages {
switch m.Role {
case "system", "developer":
if txt := m.Content.JoinText(); txt != "" {
sys = append(sys, txt)
}
case "tool":
items = append(items, map[string]any{"type": "function_call_output",
"call_id": m.ToolCallID, "output": m.Content.JoinText()})
default:
msgItems, err := chatMessageItems(m)
if err != nil {
return nil, "", err
}
items = append(items, msgItems...)
}
}
return items, strings.Join(sys, "\n\n"), nil
}
// chatMessageItems 把 user/assistant 消息转为 message 项;assistant 的
// tool_calls 追加为独立 function_call 项(内容在前,与原语序一致)。
func chatMessageItems(m aiwire.ChatMessage) ([]any, error) {
parts, err := chatContentParts(m.Role, m.Content)
if err != nil {
return nil, err
}
items := make([]any, 0, 1+len(m.ToolCalls))
if len(parts) > 0 {
items = append(items, map[string]any{"role": m.Role, "content": parts})
}
for _, tc := range m.ToolCalls {
items = append(items, map[string]any{"type": "function_call", "call_id": tc.ID,
"name": tc.Function.Name, "arguments": tc.Function.Arguments})
}
return items, nil
}
// chatContentParts 把消息内容转为 Responses 部件(文本按角色定型,图片转 input_image)。
func chatContentParts(role string, c aiwire.Content) ([]map[string]any, error) {
if !c.IsArray {
if c.Text == "" {
return nil, nil
}
return []map[string]any{respTextPart(role, c.Text)}, nil
}
parts := make([]map[string]any, 0, len(c.Parts))
for _, p := range c.Parts {
switch p.Type {
case "", "text":
parts = append(parts, respTextPart(role, p.Text))
case "image_url":
if p.ImageURL == nil || p.ImageURL.URL == "" {
return nil, fmt.Errorf("image_url 块缺少 url")
}
// detail 不透传:兼容面对该字段的支持无法证实(实测期间上游视觉请求不稳,
// 无从归因),它仅是质量提示,忽略合法且消除一个风险轴(README 披露)
parts = append(parts, map[string]any{"type": "input_image", "image_url": p.ImageURL.URL})
default:
return nil, ErrAiUnsupportedBlock
}
}
return parts, nil
}
// chatRespToolChoice 映射 tool_choice:string 形态(auto/none/required)原样;
// {"type":"function","function":{"name":N}} 拍平为 Responses 具名形态;其余忽略。
func chatRespToolChoice(raw json.RawMessage) any {
if len(raw) == 0 {
return nil
}
var s string
if json.Unmarshal(raw, &s) == nil {
switch s {
case "auto", "none", "required":
return s
}
return nil
}
var tc struct {
Type string `json:"type"`
Function struct {
Name string `json:"name"`
} `json:"function"`
}
if json.Unmarshal(raw, &tc) != nil || tc.Type != "function" || tc.Function.Name == "" {
return nil
}
return map[string]string{"type": "function", "name": tc.Function.Name}
}
// chatTextFormat 把 response_format 拍平为 text.format(json_schema 提升嵌套字段)。
func chatTextFormat(rf *aiwire.ResponseFormat) map[string]any {
if rf == nil || rf.Type == "" || rf.Type == "text" {
return nil
}
format := map[string]any{"type": rf.Type}
if rf.Type != "json_schema" || len(rf.JSONSchema) == 0 {
return format
}
var js struct {
Name string `json:"name"`
Schema json.RawMessage `json:"schema"`
Strict *bool `json:"strict"`
}
if json.Unmarshal(rf.JSONSchema, &js) != nil {
return format
}
if js.Name != "" {
format["name"] = js.Name
}
if len(js.Schema) > 0 {
format["schema"] = js.Schema
}
if js.Strict != nil {
format["strict"] = *js.Strict
}
return format
}
// ResponsesToChat 把直通非流式响应转为 Chat Completions 响应(reasoning 项丢弃)。
func ResponsesToChat(payload []byte, id string, created int64) (*aiwire.ChatResponse, error) {
var resp respPayload
if err := json.Unmarshal(payload, &resp); err != nil {
return nil, fmt.Errorf("解析上游响应: %w", err)
}
msg := aiwire.ChatMessage{Role: "assistant"}
var text strings.Builder
for _, item := range resp.Output {
switch item.Type {
case "message":
for _, part := range item.Content {
if part.Type == "output_text" {
text.WriteString(part.Text)
}
}
case "function_call":
msg.ToolCalls = append(msg.ToolCalls, aiwire.ToolCall{ID: item.CallID, Type: "function",
Function: aiwire.FunctionCall{Name: item.Name, Arguments: item.Arguments}})
}
}
msg.Content = aiwire.NewTextContent(text.String())
choice := aiwire.Choice{Index: 0, Message: msg,
FinishReason: chatFinishReason(&resp, len(msg.ToolCalls) > 0)}
return &aiwire.ChatResponse{ID: id, Object: "chat.completion", Created: created,
Model: resp.Model, Choices: []aiwire.Choice{choice}, Usage: usageFromResp(resp.Usage)}, nil
}
// chatFinishReason 映射终态:截断优先 → length;含工具调用 → tool_calls;默认 stop。
func chatFinishReason(resp *respPayload, hasTools bool) string {
if respTruncated(resp) {
return "length"
}
if hasTools {
return "tool_calls"
}
return "stop"
}
// respTruncated 报告直通响应是否因 max_output_tokens 截断。
func respTruncated(resp *respPayload) bool {
return resp != nil && resp.Status == "incomplete" && resp.IncompleteDetails != nil &&
resp.IncompleteDetails.Reason == "max_output_tokens"
}
// ---- Chat Completions 流式桥:直通 SSE 事件 → chat.completion.chunk 序列 ----
// ChatRespBridge 把直通 SSE 事件流桥接为 chunk 序列:首个增量补 role,文本与
// 工具增量按 OpenAI 形态输出;reasoning 系列事件丢弃;[DONE] 由传输层写出。
type ChatRespBridge struct {
id, model string
created int64
includeUsage bool
started bool
toolIndex int
inTool bool
finish string
usage *aiwire.Usage
}
// NewChatRespBridge 构造桥;includeUsage 即 stream_options.include_usage。
func NewChatRespBridge(id, model string, created int64, includeUsage bool) *ChatRespBridge {
return &ChatRespBridge{id: id, model: model, created: created,
includeUsage: includeUsage, toolIndex: -1, finish: "stop"}
}
// Feed 消费一行 SSE data JSON,返回应立即写出的 chunk。
func (b *ChatRespBridge) Feed(data []byte) []aiwire.ChatChunk {
var ev respStreamEvent
if json.Unmarshal(data, &ev) != nil {
return nil
}
switch ev.Type {
case "response.output_text.delta":
b.inTool = false
return []aiwire.ChatChunk{b.chunk(b.deltaWithRole(aiwire.Delta{Content: ev.Delta}), nil)}
case "response.output_item.added":
return b.toolOpen(ev.Item)
case "response.function_call_arguments.delta":
return b.argsChunk(ev.Delta)
case "response.completed", "response.incomplete", "response.failed":
b.finishFrom(ev.Response)
}
return nil
}
// chunk 装配一条单 choice 增量事件。
func (b *ChatRespBridge) chunk(delta aiwire.Delta, finish *string) aiwire.ChatChunk {
return aiwire.ChatChunk{ID: b.id, Object: "chat.completion.chunk", Created: b.created,
Model: b.model, Choices: []aiwire.ChunkChoice{{Index: 0, Delta: delta, FinishReason: finish}}}
}
// deltaWithRole 给首个增量补 role(OpenAI 首块携带 role 语义)。
func (b *ChatRespBridge) deltaWithRole(d aiwire.Delta) aiwire.Delta {
if !b.started {
b.started = true
d.Role = "assistant"
}
return d
}
// toolOpen 在新 function_call 输出项开始时发 id/name 增量并推进聚合 index;
// 非工具输出项只复位增量目标。
func (b *ChatRespBridge) toolOpen(item *respOutputItem) []aiwire.ChatChunk {
if item == nil || item.Type != "function_call" {
b.inTool = false
return nil
}
b.toolIndex++
b.inTool = true
b.finish = "tool_calls"
d := aiwire.Delta{ToolCalls: []aiwire.ToolCallDelta{{Index: b.toolIndex, ID: item.CallID,
Type: "function", Function: aiwire.FunctionCallDelta{Name: item.Name}}}}
return []aiwire.ChatChunk{b.chunk(b.deltaWithRole(d), nil)}
}
// argsChunk 输出当前工具的实参增量(仅在工具输出项进行中)。
func (b *ChatRespBridge) argsChunk(delta string) []aiwire.ChatChunk {
if !b.inTool {
return nil
}
d := aiwire.Delta{ToolCalls: []aiwire.ToolCallDelta{{Index: b.toolIndex,
Function: aiwire.FunctionCallDelta{Arguments: delta}}}}
return []aiwire.ChatChunk{b.chunk(d, nil)}
}
// finishFrom 记录终态 usage 与截断语义(max_output_tokens → length)。
func (b *ChatRespBridge) finishFrom(resp *respPayload) {
if resp == nil {
return
}
b.usage = usageFromResp(resp.Usage)
if respTruncated(resp) {
b.finish = "length"
}
}
// Finish 在上游流结束后收尾:终块带 finish_reason;include_usage 时追加 usage 块
// (choices 为空数组,OpenAI 规范形态;上游未给 usage 时输出零值)。
func (b *ChatRespBridge) Finish() []aiwire.ChatChunk {
chunks := []aiwire.ChatChunk{b.chunk(b.deltaWithRole(aiwire.Delta{}), &b.finish)}
if !b.includeUsage {
return chunks
}
usage := b.usage
if usage == nil {
usage = &aiwire.Usage{}
}
return append(chunks, aiwire.ChatChunk{ID: b.id, Object: "chat.completion.chunk",
Created: b.created, Model: b.model, Choices: []aiwire.ChunkChoice{}, Usage: usage})
}
// Usage 返回聚合到的用量(供调用日志),上游未报告时为 nil。
func (b *ChatRespBridge) Usage() *aiwire.Usage { return b.usage }
internal/service/chatresponses_test.go
+301
View File
@@ -0,0 +1,301 @@
package service
import (
"encoding/json"
"strings"
"testing"
"oci-portal/internal/aiwire"
)
func mustChatReq(t *testing.T, raw string) aiwire.ChatRequest {
t.Helper()
var req aiwire.ChatRequest
if err := json.Unmarshal([]byte(raw), &req); err != nil {
t.Fatalf("解析请求: %v", err)
}
return req
}
func mustChatBody(t *testing.T, raw string) map[string]any {
t.Helper()
payload, err := ChatToResponsesBody(mustChatReq(t, raw))
if err != nil {
t.Fatalf("ChatToResponsesBody: %v", err)
}
var body map[string]any
if err := json.Unmarshal(payload, &body); err != nil {
t.Fatalf("unmarshal: %v", err)
}
return body
}
// TestChatToResponsesBody 断言消息展开、instructions 聚合、工具与顶层字段装配。
func TestChatToResponsesBody(t *testing.T) {
body := mustChatBody(t, `{
"model": "xai.grok-4.3", "max_tokens": 64, "max_completion_tokens": 128,
"temperature": 0.5, "top_p": 0.9, "parallel_tool_calls": false, "stream": true,
"reasoning_effort": "HIGH",
"messages": [
{"role": "system", "content": "你是助手"},
{"role": "developer", "content": [{"type": "text", "text": "简洁作答"}]},
{"role": "user", "content": "东京天气?"},
{"role": "assistant", "content": "查询中", "tool_calls": [
{"id": "t1", "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\":\"东京\"}"}}
]},
{"role": "tool", "tool_call_id": "t1", "content": "晴 25 度"},
{"role": "user", "content": "继续"}
],
"tools": [{"type": "function", "function": {"name": "get_weather", "description": "查天气", "parameters": {"type": "object"}}}],
"tool_choice": "required"
}`)
if body["model"] != "xai.grok-4.3" || body["store"] != false || body["stream"] != true {
t.Fatalf("顶层字段装配错误: %v", body)
}
if body["max_output_tokens"] != float64(128) {
t.Fatalf("max_output_tokens = %v, want 128(max_completion_tokens 优先)", body["max_output_tokens"])
}
if body["instructions"] != "你是助手\n\n简洁作答" {
t.Fatalf("instructions = %v", body["instructions"])
}
if body["temperature"] != 0.5 || body["top_p"] != 0.9 || body["parallel_tool_calls"] != false {
t.Fatalf("采样字段装配错误: %v", body)
}
if body["tool_choice"] != "required" {
t.Fatalf("tool_choice = %v", body["tool_choice"])
}
if reasoning, _ := body["reasoning"].(map[string]any); reasoning["effort"] != "high" {
t.Fatalf("effort = %v, want high(小写透传)", body["reasoning"])
}
kinds := chatInputKinds(t, body)
want := "message/user,message/assistant,function_call,function_call_output,message/user"
if kinds != want {
t.Fatalf("input 顺序 = %s, want %s", kinds, want)
}
tools, _ := body["tools"].([]any)
tool, _ := tools[0].(map[string]any)
if tool["type"] != "function" || tool["name"] != "get_weather" {
t.Fatalf("工具应拍平为 Responses 形态: %v", tool)
}
}
func chatInputKinds(t *testing.T, body map[string]any) string {
t.Helper()
input, _ := body["input"].([]any)
kinds := make([]string, 0, len(input))
for _, it := range input {
m := it.(map[string]any)
if ty, ok := m["type"].(string); ok {
kinds = append(kinds, ty)
} else {
kinds = append(kinds, "message/"+m["role"].(string))
}
}
return strings.Join(kinds, ",")
}
// TestChatToResponsesBodyContent 断言文本部件按角色定型与图片装配/拒绝。
func TestChatToResponsesBodyContent(t *testing.T) {
body := mustChatBody(t, `{"model":"m","messages":[
{"role":"user","content":[
{"type":"text","text":"看图"},
{"type":"image_url","image_url":{"url":"data:image/png;base64,QUJD"}}]},
{"role":"assistant","content":"这是猫"}]}`)
raw, _ := json.Marshal(body["input"])
if !strings.Contains(string(raw), `"input_image"`) ||
!strings.Contains(string(raw), "data:image/png;base64,QUJD") {
t.Fatalf("图片应转 input_image: %s", raw)
}
if !strings.Contains(string(raw), `"output_text"`) {
t.Fatalf("assistant 历史文本应为 output_text 部件: %s", raw)
}
for name, msg := range map[string]string{
"audio 块": `{"role":"user","content":[{"type":"input_audio","input_audio":{}}]}`,
"图片缺 url": `{"role":"user","content":[{"type":"image_url","image_url":{}}]}`,
} {
raw := `{"model":"m","messages":[` + msg + `]}`
if _, err := ChatToResponsesBody(mustChatReq(t, raw)); err == nil {
t.Errorf("%s 应拒绝", name)
}
}
badTool := mustChatReq(t, `{"model":"m","messages":[{"role":"user","content":"hi"}],
"tools":[{"type":"web_search","function":{}}]}`)
if _, err := ChatToResponsesBody(badTool); err == nil {
t.Error("非 function 工具类型应拒绝")
}
}
// TestChatRespToolChoice 断言 tool_choice 各形态映射。
func TestChatRespToolChoice(t *testing.T) {
cases := []struct {
name, raw string
want any
}{
{"auto", `"auto"`, "auto"},
{"none", `"none"`, "none"},
{"required", `"required"`, "required"},
{"具名 function", `{"type":"function","function":{"name":"f1"}}`,
map[string]string{"type": "function", "name": "f1"}},
{"未知 string", `"whatever"`, nil},
{"缺 name", `{"type":"function","function":{}}`, nil},
{"空", ``, nil},
}
for _, tc := range cases {
got := chatRespToolChoice(json.RawMessage(tc.raw))
if gotMap, ok := got.(map[string]string); ok {
wantMap, _ := tc.want.(map[string]string)
if wantMap == nil || gotMap["name"] != wantMap["name"] {
t.Errorf("%s: got %v, want %v", tc.name, got, tc.want)
}
continue
}
if got != tc.want {
t.Errorf("%s: got %v, want %v", tc.name, got, tc.want)
}
}
}
// TestChatTextFormat 断言 response_format 拍平映射。
func TestChatTextFormat(t *testing.T) {
if got := chatTextFormat(nil); got != nil {
t.Errorf("nil 应不下发: %v", got)
}
if got := chatTextFormat(&aiwire.ResponseFormat{Type: "text"}); got != nil {
t.Errorf("text 应不下发: %v", got)
}
if got := chatTextFormat(&aiwire.ResponseFormat{Type: "json_object"}); got["type"] != "json_object" {
t.Errorf("json_object: %v", got)
}
got := chatTextFormat(&aiwire.ResponseFormat{Type: "json_schema",
JSONSchema: json.RawMessage(`{"name":"out","schema":{"type":"object"},"strict":true}`)})
if got["type"] != "json_schema" || got["name"] != "out" || got["strict"] != true {
t.Errorf("json_schema 应提升嵌套字段: %v", got)
}
if _, ok := got["schema"]; !ok {
t.Errorf("schema 缺失: %v", got)
}
}
// TestResponsesToChat 断言文本聚合、tool_calls、finish_reason 与 usage 回转。
func TestResponsesToChat(t *testing.T) {
payload := []byte(`{"model":"xai.grok-4.3","status":"completed","output":[
{"type":"reasoning","summary":[]},
{"type":"message","content":[{"type":"output_text","text":"你"},{"type":"output_text","text":"好"}]},
{"type":"function_call","call_id":"c1","name":"get_weather","arguments":"{\"city\":\"东京\"}"}],
"usage":{"input_tokens":10,"output_tokens":5,"total_tokens":15,"input_tokens_details":{"cached_tokens":4}}}`)
out, err := ResponsesToChat(payload, "chatcmpl-1", 1700000000)
if err != nil {
t.Fatalf("ResponsesToChat: %v", err)
}
if out.Object != "chat.completion" || out.ID != "chatcmpl-1" || out.Created != 1700000000 {
t.Fatalf("响应骨架错误: %+v", out)
}
msg := out.Choices[0].Message
if msg.Role != "assistant" || msg.Content.JoinText() != "你好" {
t.Fatalf("content 装配错误: %+v", msg)
}
if len(msg.ToolCalls) != 1 || msg.ToolCalls[0].ID != "c1" ||
msg.ToolCalls[0].Function.Arguments != `{"city":"东京"}` {
t.Fatalf("tool_calls 装配错误: %+v", msg.ToolCalls)
}
if out.Choices[0].FinishReason != "tool_calls" {
t.Fatalf("finish_reason = %s, want tool_calls", out.Choices[0].FinishReason)
}
if out.Usage.PromptTokens != 10 || out.Usage.CompletionTokens != 5 || out.Usage.CachedTokens() != 4 {
t.Fatalf("usage = %+v", out.Usage)
}
trunc := []byte(`{"model":"m","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},
"output":[{"type":"message","content":[{"type":"output_text","text":"半"}]}]}`)
out2, err := ResponsesToChat(trunc, "chatcmpl-2", 1)
if err != nil || out2.Choices[0].FinishReason != "length" {
t.Fatalf("截断 finish_reason = %+v, %v", out2.Choices, err)
}
if out2.Usage != nil {
t.Fatalf("无 usage 时应为 nil: %+v", out2.Usage)
}
}
func chatChunkShapes(chunks []aiwire.ChatChunk) string {
kinds := make([]string, 0, len(chunks))
for _, ch := range chunks {
switch {
case len(ch.Choices) == 0:
kinds = append(kinds, "usage")
case ch.Choices[0].FinishReason != nil:
kinds = append(kinds, "finish:"+*ch.Choices[0].FinishReason)
case len(ch.Choices[0].Delta.ToolCalls) > 0:
kinds = append(kinds, "tool")
case ch.Choices[0].Delta.Role != "":
kinds = append(kinds, "role+text")
default:
kinds = append(kinds, "text")
}
}
return strings.Join(kinds, ",")
}
// TestChatRespBridge 断言流桥:首块 role、文本与工具增量、reasoning 丢弃、usage 块。
func TestChatRespBridge(t *testing.T) {
b := NewChatRespBridge("chatcmpl-1", "m1", 1700000000, true)
var chunks []aiwire.ChatChunk
feed := func(lines ...string) {
for _, l := range lines {
chunks = append(chunks, b.Feed([]byte(l))...)
}
}
feed(`{"type":"response.created","response":{"model":"m1"}}`,
`{"type":"response.reasoning_text.delta","delta":"思考中"}`,
`{"type":"response.output_text.delta","delta":"你"}`,
`{"type":"response.output_text.delta","delta":"好"}`,
`{"type":"response.output_item.added","item":{"type":"function_call","call_id":"c1","name":"f"}}`,
`{"type":"response.function_call_arguments.delta","delta":"{\"a\":"}`,
`{"type":"response.function_call_arguments.delta","delta":"1}"}`,
`{"type":"response.completed","response":{"status":"completed","usage":{"input_tokens":8,"output_tokens":4,"total_tokens":12}}}`)
chunks = append(chunks, b.Finish()...)
got := chatChunkShapes(chunks)
want := "role+text,text,tool,tool,tool,finish:tool_calls,usage"
if got != want {
t.Fatalf("chunk 序列:\n got %s\nwant %s", got, want)
}
if chunks[0].Choices[0].Delta.Role != "assistant" || chunks[0].Choices[0].Delta.Content != "你" {
t.Fatalf("首块应含 role 与文本: %+v", chunks[0].Choices[0].Delta)
}
tool := chunks[2].Choices[0].Delta.ToolCalls[0]
if tool.Index != 0 || tool.ID != "c1" || tool.Function.Name != "f" {
t.Fatalf("工具首块 = %+v", tool)
}
args := chunks[3].Choices[0].Delta.ToolCalls[0].Function.Arguments +
chunks[4].Choices[0].Delta.ToolCalls[0].Function.Arguments
if args != `{"a":1}` {
t.Fatalf("实参增量聚合 = %s", args)
}
last := chunks[len(chunks)-1]
if last.Usage == nil || last.Usage.PromptTokens != 8 || len(last.Choices) != 0 {
t.Fatalf("usage 块 = %+v", last)
}
if b.Usage().TotalTokens != 12 {
t.Fatalf("Usage() = %+v", b.Usage())
}
}
// TestChatRespBridgeVariants 断言 include_usage 关闭、截断与空流的收尾形态。
func TestChatRespBridgeVariants(t *testing.T) {
noUsage := NewChatRespBridge("c", "m", 1, false)
noUsage.Feed([]byte(`{"type":"response.output_text.delta","delta":"x"}`))
noUsage.Feed([]byte(`{"type":"response.incomplete","response":{"status":"incomplete","incomplete_details":{"reason":"max_output_tokens"}}}`))
if got := chatChunkShapes(noUsage.Finish()); got != "finish:length" {
t.Errorf("截断且不带 usage 块: %s", got)
}
empty := NewChatRespBridge("c", "m", 1, false)
fin := empty.Finish()
if got := chatChunkShapes(fin); got != "finish:stop" {
t.Errorf("空流收尾 = %s", got)
}
if fin[0].Choices[0].Delta.Role != "assistant" {
t.Errorf("空流终块应补 role: %+v", fin[0].Choices[0].Delta)
}
}
internal/service/logevent.go
+121 -24
View File
@@ -57,9 +57,6 @@ type LogEventService struct {
relayPollTick time.Duration // 订阅确认轮询间隔,零值用默认
relayPollTimeout time.Duration // 订阅确认轮询上限,零值用默认
alertMu sync.Mutex // 保护告警规则冷却表
alertSentAt map[uint]time.Time // 规则 ID → 上次告警时刻(阈值型规则冷却)
}
// NewLogEventService 组装依赖;调用 StartParser / StartCleanup 后台协程后生效。
@@ -373,20 +370,18 @@ func (s *LogEventService) parseOnce(ctx context.Context) {
if len(events) == 0 {
return
}
rules := s.loadEnabledAlertRules(ctx)
for i := range events {
s.processLogEvent(ctx, &events[i], rules)
s.processLogEvent(ctx, &events[i])
}
}
// processLogEvent 仅在条件更新命中原行后触发通知,删除并发胜出时静默跳过。
func (s *LogEventService) processLogEvent(ctx context.Context, event *model.LogEvent, rules []model.AlertRule) {
func (s *LogEventService) processLogEvent(ctx context.Context, event *model.LogEvent) {
parsed := parseLogEvent([]byte(event.Payload))
if !s.updateParsedEvent(ctx, event, parsed) {
return
}
s.notifyCritical(ctx, event, parsed)
s.matchAlertRules(ctx, rules, event, parsed)
}
// updateParsedEvent 用条件 UPDATE 禁止 Save 在删除后隐式重建事件。
@@ -413,26 +408,49 @@ func (s *LogEventService) updateParsedEvent(ctx context.Context, event *model.Lo
// onsEnvelope 覆盖 ONS 消息与 CloudEvents 审计事件的常见字段;
// 真实格式以联调实测为准,提不出字段时只置 Processed 不回填。
type onsEnvelope struct {
EventType string `json:"eventType"`
Type string `json:"type"`
Source string `json:"source"`
EventTime string `json:"eventTime"`
ResourceName string `json:"resourceName"`
Data json.RawMessage `json:"data"`
Identity *onsIdentity `json:"identity"`
EventType string `json:"eventType"`
Type string `json:"type"`
Source string `json:"source"`
EventTime string `json:"eventTime"`
ResourceName string `json:"resourceName"`
Message string `json:"message"`
Data json.RawMessage `json:"data"`
Identity *onsIdentity `json:"identity"`
AdditionalDetails *onsAddDetails `json:"additionalDetails"`
StateChange *onsStateChange `json:"stateChange"`
}
// onsIdentity 是 Audit 事件 data.identity 中与展示相关的字段。
type onsIdentity struct {
IPAddress string `json:"ipAddress"`
IPAddress string `json:"ipAddress"`
PrincipalName string `json:"principalName"`
}
// parsedEvent 是从消息原文提取的展示字段集;ResourceName 仅供 P2 推送文案,不落库。
// onsAddDetails 是 IDCS 登录类事件 data.additionalDetails 的补充字段;
// AuditEventMapValue 为嵌套的 JSON 字符串(含 eventId 成败与失败原因)。
type onsAddDetails struct {
ActorName string `json:"actorName"`
ClientIP string `json:"clientIp"`
AuditEventMapValue string `json:"auditEventMapValue"`
}
// onsStateChange 承载 Audit v2 的资源变更快照;description 供策略类事件推送文案。
type onsStateChange struct {
Current struct {
Description string `json:"description"`
} `json:"current"`
}
// parsedEvent 是从消息原文提取的展示字段集;ResourceName/Actor/Outcome/Detail
// 仅供 P2 推送文案,不落库。
type parsedEvent struct {
EventType string
Source string
SourceIP string
ResourceName string
Actor string // 操作者(identity.principalName,登录事件回退 actorName)
Outcome string // 成功 / 失败 / 空(判读不出)
Detail string // 补充说明(策略描述、登录失败原因等)
EventTime *time.Time
}
@@ -455,7 +473,7 @@ func parseLogEvent(payload []byte) parsedEvent {
return envelopeFields(env)
}
// mergeEnvelope 外层缺失字段时以 data 内层补齐(Audit 的 identity 在内层)。
// mergeEnvelope 外层缺失字段时以 data 内层补齐(Audit 的 identity 等在内层)。
func mergeEnvelope(outer, inner onsEnvelope) onsEnvelope {
if outer.EventType == "" {
outer.EventType = inner.EventType
@@ -472,13 +490,22 @@ func mergeEnvelope(outer, inner onsEnvelope) onsEnvelope {
if outer.ResourceName == "" {
outer.ResourceName = inner.ResourceName
}
if outer.Message == "" {
outer.Message = inner.Message
}
if outer.Identity == nil {
outer.Identity = inner.Identity
}
if outer.AdditionalDetails == nil {
outer.AdditionalDetails = inner.AdditionalDetails
}
if outer.StateChange == nil {
outer.StateChange = inner.StateChange
}
return outer
}
// envelopeFields 收敛字段别名并解析事件时间。
// envelopeFields 收敛字段别名并解析事件时间与推送用补充字段。
func envelopeFields(env onsEnvelope) parsedEvent {
eventType := env.EventType
if eventType == "" {
@@ -490,19 +517,81 @@ func envelopeFields(env onsEnvelope) parsedEvent {
eventTime = &ts
}
}
ip := ""
if env.Identity != nil {
ip = env.Identity.IPAddress
}
outcome, detail := envOutcome(env)
return parsedEvent{
EventType: clip(eventType, 128),
Source: clip(env.Source, 64),
SourceIP: clip(ip, 64),
SourceIP: clip(envIP(env), 64),
ResourceName: clip(env.ResourceName, 128),
Actor: clipRunes(envActor(env), 64),
Outcome: outcome,
Detail: clipRunes(detail, 200),
EventTime: eventTime,
}
}
// envIP 取发起方 IP:Audit 的 identity.ipAddress,登录事件回退 additionalDetails.clientIp。
func envIP(env onsEnvelope) string {
if env.Identity != nil && env.Identity.IPAddress != "" {
return env.Identity.IPAddress
}
if env.AdditionalDetails != nil {
return env.AdditionalDetails.ClientIP
}
return ""
}
// envActor 取操作者:Audit 的 identity.principalName,登录事件回退 additionalDetails.actorName。
func envActor(env onsEnvelope) string {
if env.Identity != nil && env.Identity.PrincipalName != "" {
return env.Identity.PrincipalName
}
if env.AdditionalDetails != nil {
return env.AdditionalDetails.ActorName
}
return ""
}
// envOutcome 判读事件成败与补充说明:登录事件解析 auditEventMapValue;其余
// Audit 事件看 message 后缀,补充说明取 stateChange.current.description(策略描述)。
func envOutcome(env onsEnvelope) (outcome, detail string) {
if env.StateChange != nil {
detail = env.StateChange.Current.Description
}
if env.AdditionalDetails != nil && env.AdditionalDetails.AuditEventMapValue != "" {
return ssoOutcome(env.AdditionalDetails.AuditEventMapValue, detail)
}
switch {
case strings.HasSuffix(env.Message, " succeeded"):
outcome = "成功"
case strings.HasSuffix(env.Message, " failed"):
outcome = "失败"
}
return outcome, detail
}
// ssoOutcome 解析 IDCS 审计负载(JSON 字符串):eventId 含 success/failure 定成败,
// 失败时以其 message 作为原因说明。
func ssoOutcome(raw, detail string) (string, string) {
var ev struct {
EventID string `json:"eventId"`
Message string `json:"message"`
}
if json.Unmarshal([]byte(raw), &ev) != nil {
return "", detail
}
switch {
case strings.Contains(ev.EventID, "success"):
return "成功", detail
case strings.Contains(ev.EventID, "failure"):
if ev.Message != "" {
detail = ev.Message
}
return "失败", detail
}
return "", detail
}
// clip 按模型列宽截断解析出的字段。
func clip(s string, max int) string {
if len(s) > max {
@@ -511,6 +600,15 @@ func clip(s string, max int) string {
return s
}
// clipRunes 按字符数截断(通知文案用,中文安全)。
func clipRunes(s string, max int) string {
r := []rune(s)
if len(r) > max {
return string(r[:max])
}
return s
}
// StartCleanup 启动周期清理:启动即清一次,之后每 24h 一次,随 ctx 取消退出。
func (s *LogEventService) StartCleanup(ctx context.Context) {
s.wg.Add(1)
@@ -535,7 +633,6 @@ func (s *LogEventService) cleanupOnce(ctx context.Context) {
if err := s.cleanup(ctx, logEventRetention, logEventMaxRows); err != nil {
log.Printf("log event cleanup: %v", err)
}
s.cleanupAlertHits(ctx)
}
// cleanup 先删过期记录,再对超量部分删最旧;阈值参数化便于测试。
internal/service/logevent_test.go
+37 -8
View File
@@ -30,8 +30,7 @@ func newLogEventEnv(t *testing.T) (*LogEventService, *gorm.DB, uint) {
t.Fatalf("db handle: %v", err)
}
sqlDB.SetMaxOpenConns(1)
if err := db.AutoMigrate(&model.Setting{}, &model.OciConfig{}, &model.LogEvent{},
&model.AlertRule{}, &model.AlertRuleHit{}); err != nil {
if err := db.AutoMigrate(&model.Setting{}, &model.OciConfig{}, &model.LogEvent{}); err != nil {
t.Fatalf("auto migrate: %v", err)
}
cfg := model.OciConfig{Alias: "测试租户"}
@@ -240,12 +239,15 @@ func TestIngestRejectsUnknownConfig(t *testing.T) {
func TestParseLogEvent(t *testing.T) {
ts := "2026-07-07T08:00:00Z"
tests := []struct {
name string
payload string
wantType string
wantSource string
wantIP string
wantTime bool
name string
payload string
wantType string
wantSource string
wantIP string
wantTime bool
wantActor string
wantOutcome string
wantDetail string
}{
{
name: "CloudEvents 单事件",
@@ -283,6 +285,29 @@ func TestParseLogEvent(t *testing.T) {
payload: `{"eventType":"x","data":{"identity":{}}}`,
wantType: "x",
},
{
name: "Audit v2 提取操作者成败与策略描述",
payload: `{"eventType":"com.oraclecloud.identityControlPlane.CreatePolicy","data":{
"identity":{"principalName":"Alfonso Garcia","ipAddress":"129.159.43.9"},
"message":"ociportal-logs-sch CreatePolicy succeeded",
"stateChange":{"current":{"description":"允许 Connector 发布到 ONS"}}}}`,
wantType: "com.oraclecloud.identityControlPlane.CreatePolicy", wantIP: "129.159.43.9",
wantActor: "Alfonso Garcia", wantOutcome: "成功", wantDetail: "允许 Connector 发布到 ONS",
},
{
name: "IDCS 登录失败提取用户 IP 与原因",
payload: `{"eventType":"com.oraclecloud.IdentitySignOn.InteractiveLogin","data":{
"additionalDetails":{"actorName":"oci","clientIp":"155.117.82.111",
"auditEventMapValue":"{\"eventId\":\"sso.authentication.failure\",\"message\":\"Authentication failure : incorrect password.\"}"}}}`,
wantType: "com.oraclecloud.IdentitySignOn.InteractiveLogin", wantIP: "155.117.82.111",
wantActor: "oci", wantOutcome: "失败", wantDetail: "Authentication failure : incorrect password.",
},
{
name: "message failed 后缀判失败",
payload: `{"eventType":"x","data":{"message":"vm TerminateInstance failed"}}`,
wantType: "x",
wantOutcome: "失败",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -291,6 +316,10 @@ func TestParseLogEvent(t *testing.T) {
t.Errorf("parse = (%q,%q,%q), want (%q,%q,%q)",
got.EventType, got.Source, got.SourceIP, tt.wantType, tt.wantSource, tt.wantIP)
}
if got.Actor != tt.wantActor || got.Outcome != tt.wantOutcome || got.Detail != tt.wantDetail {
t.Errorf("推送字段 = (%q,%q,%q), want (%q,%q,%q)",
got.Actor, got.Outcome, got.Detail, tt.wantActor, tt.wantOutcome, tt.wantDetail)
}
if (got.EventTime != nil) != tt.wantTime {
t.Errorf("eventTime present = %v, want %v", got.EventTime != nil, tt.wantTime)
}
internal/service/logrelay.go
+23 -1
View File
@@ -322,12 +322,34 @@ func relayEventShortName(eventType string) string {
}
// criticalEventVars 判定关键事件并生成模板变量;非关键事件 ok 为 false。
// actor/ip/outcome/resource 缺失时兜底 —,detail 包装为独立行(空则不占行)。
func criticalEventVars(alias string, p parsedEvent) (map[string]string, bool) {
name := relayEventShortName(p.EventType)
if name == "" || !relayCriticalSet[name] {
return nil, false
}
return map[string]string{"tenant": alias, "event": name, "resource": p.ResourceName}, true
return map[string]string{
"tenant": alias, "event": name,
"resource": orDash(p.ResourceName), "actor": orDash(p.Actor),
"ip": orDash(p.SourceIP), "outcome": orDash(p.Outcome),
"detail": detailLine(p.Detail),
}, true
}
// orDash 空值兜底为 —,避免模板出现悬空标签。
func orDash(v string) string {
if v == "" {
return "—"
}
return v
}
// detailLine 把补充说明包装为独立行;为空时不产生多余空行。
func detailLine(v string) string {
if v == "" {
return ""
}
return "\n" + v
}
// notifyCritical 对关键事件推送告警;依赖未注入或对应子类开关关闭时跳过。
internal/service/logrelay_test.go
+32 -20
View File
@@ -313,26 +313,36 @@ func TestRelayReady(t *testing.T) {
func TestCriticalEventText(t *testing.T) {
tests := []struct {
name string
event parsedEvent
wantOK bool
wantText string
name string
event parsedEvent
wantOK bool
want map[string]string
}{
{
name: "实例终止命中",
event: parsedEvent{EventType: "com.oraclecloud.ComputeApi.TerminateInstance", ResourceName: "vm-1"},
wantOK: true,
wantText: "☁️ 云端事件:免费01 TerminateInstance vm-1",
name: "实例终止含操作者与成败",
event: parsedEvent{EventType: "com.oraclecloud.ComputeApi.TerminateInstance",
ResourceName: "vm-1", Actor: "demo@example.com", SourceIP: "203.0.113.8", Outcome: "成功"},
wantOK: true,
want: map[string]string{"event": "TerminateInstance", "resource": "vm-1",
"actor": "demo@example.com", "ip": "203.0.113.8", "outcome": "成功", "detail": ""},
},
{
name: "无资源名省略尾段",
event: parsedEvent{EventType: "com.oraclecloud.IdentityControlPlane.CreateApiKey"},
wantOK: true,
wantText: "☁️ 云端事件:免费01 CreateApiKey",
name: "字段缺失兜底破折号",
event: parsedEvent{EventType: "com.oraclecloud.IdentityControlPlane.CreateApiKey"},
wantOK: true,
want: map[string]string{"event": "CreateApiKey", "resource": "—",
"actor": "—", "ip": "—", "outcome": "—", "detail": ""},
},
{
name: "补充说明独立成行",
event: parsedEvent{EventType: "CreatePolicy", Detail: "允许发布到 ONS Topic"},
wantOK: true,
want: map[string]string{"event": "CreatePolicy", "detail": "\n允许发布到 ONS Topic"},
},
{name: "List 噪声不推", event: parsedEvent{EventType: "com.oraclecloud.ComputeApi.ListInstances"}, wantOK: false},
{name: "空类型不推", event: parsedEvent{}, wantOK: false},
{name: "短名直接命中", event: parsedEvent{EventType: "LaunchInstance"}, wantOK: true, wantText: "☁️ 云端事件:免费01 LaunchInstance"},
{name: "短名直接命中", event: parsedEvent{EventType: "LaunchInstance"}, wantOK: true,
want: map[string]string{"event": "LaunchInstance"}},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -340,13 +350,15 @@ func TestCriticalEventText(t *testing.T) {
if ok != tt.wantOK {
t.Fatalf("ok = %v, want %v", ok, tt.wantOK)
}
if ok {
got := "☁️ 云端事件:" + vars["tenant"] + " " + vars["event"]
if vars["resource"] != "" {
got += " " + vars["resource"]
}
if got != tt.wantText {
t.Errorf("vars 渲染 = %q, want %q", got, tt.wantText)
if !ok {
return
}
if vars["tenant"] != "免费01" {
t.Errorf("tenant = %q, want 免费01", vars["tenant"])
}
for k, want := range tt.want {
if vars[k] != want {
t.Errorf("%s = %q, want %q", k, vars[k], want)
}
}
})
internal/service/notifytpl.go
+22 -23
View File
@@ -21,7 +21,6 @@ var notifyTplOrder = []string{
"task_fail", "task_recover", "snatch_success", "tenant_dead", "task_stop",
"login_lock", "model_deprecated",
"log_event_instance", "log_event_identity", "log_event_policy", "log_event_region", "log_event_login",
"audit_alert",
}
var notifyTplDefs = map[string]notifyTplDef{
@@ -73,40 +72,40 @@ var notifyTplDefs = map[string]notifyTplDef{
},
"log_event_instance": {
Label: "实例生命周期",
Default: "☁️ 云端事件:{{tenant}} {{event}} {{resource}}",
Vars: []string{"tenant", "event", "resource"},
Sample: map[string]string{"tenant": "免费01", "event": "TerminateInstance", "resource": "web-server-1"},
Default: "☁️ 云端事件:{{tenant}} {{event}} {{resource}} · {{outcome}}\n操作者 {{actor}} · 来源 {{ip}}{{detail}}",
Vars: []string{"tenant", "event", "resource", "actor", "ip", "outcome", "detail"},
Sample: map[string]string{"tenant": "免费01", "event": "TerminateInstance", "resource": "web-server-1",
"actor": "demo@example.com", "ip": "203.0.113.8", "outcome": "成功", "detail": ""},
},
"log_event_identity": {
Label: "用户与凭据",
Default: "☁️ 云端事件:{{tenant}} {{event}} {{resource}}",
Vars: []string{"tenant", "event", "resource"},
Sample: map[string]string{"tenant": "免费01", "event": "CreateApiKey", "resource": "user/demo"},
Default: "☁️ 云端事件:{{tenant}} {{event}} {{resource}} · {{outcome}}\n操作者 {{actor}} · 来源 {{ip}}{{detail}}",
Vars: []string{"tenant", "event", "resource", "actor", "ip", "outcome", "detail"},
Sample: map[string]string{"tenant": "免费01", "event": "CreateApiKey", "resource": "user/demo",
"actor": "admin@example.com", "ip": "203.0.113.8", "outcome": "成功", "detail": ""},
},
"log_event_policy": {
Label: "策略变更",
Default: "☁️ 云端事件:{{tenant}} {{event}} {{resource}}",
Vars: []string{"tenant", "event", "resource"},
Sample: map[string]string{"tenant": "免费01", "event": "UpdatePolicy", "resource": "admin-policy"},
Default: "☁️ 云端事件:{{tenant}} {{event}} {{resource}} · {{outcome}}\n操作者 {{actor}} · 来源 {{ip}}{{detail}}",
Vars: []string{"tenant", "event", "resource", "actor", "ip", "outcome", "detail"},
Sample: map[string]string{"tenant": "免费01", "event": "CreatePolicy", "resource": "admin-policy",
"actor": "admin@example.com", "ip": "203.0.113.8", "outcome": "成功",
"detail": "\n允许 Service Connector 发布到 ONS Topic"},
},
"log_event_region": {
Label: "区域订阅",
Default: "☁️ 云端事件:{{tenant}} {{event}} {{resource}}",
Vars: []string{"tenant", "event", "resource"},
Sample: map[string]string{"tenant": "免费01", "event": "CreateRegionSubscription", "resource": "ap-osaka-1"},
Default: "☁️ 云端事件:{{tenant}} {{event}} {{resource}} · {{outcome}}\n操作者 {{actor}} · 来源 {{ip}}{{detail}}",
Vars: []string{"tenant", "event", "resource", "actor", "ip", "outcome", "detail"},
Sample: map[string]string{"tenant": "免费01", "event": "CreateRegionSubscription", "resource": "ap-osaka-1",
"actor": "admin@example.com", "ip": "203.0.113.8", "outcome": "成功", "detail": ""},
},
"log_event_login": {
Label: "控制台登录",
Default: "☁️ 云端事件:{{tenant}} {{event}} {{resource}}",
Vars: []string{"tenant", "event", "resource"},
Sample: map[string]string{"tenant": "免费01", "event": "InteractiveLogin", "resource": "demo@example.com"},
},
"audit_alert": {
Label: "审计告警",
Default: "🔔 审计告警:{{rule}}\n租户 {{tenant}} · {{event}} {{resource}}\n来源 {{ip}} · 窗口内 {{count}} 次",
Vars: []string{"rule", "tenant", "event", "resource", "ip", "count"},
Sample: map[string]string{"rule": "非白名单终止实例", "tenant": "免费01",
"event": "TerminateInstance", "resource": "web-server-1", "ip": "203.0.113.8", "count": "1"},
Default: "☁️ 云端事件:{{tenant}} {{event}} · {{outcome}}\n用户 {{actor}} · 登录 IP {{ip}}{{detail}}",
Vars: []string{"tenant", "event", "resource", "actor", "ip", "outcome", "detail"},
Sample: map[string]string{"tenant": "免费01", "event": "InteractiveLogin", "resource": "—",
"actor": "demo@example.com", "ip": "155.117.82.111", "outcome": "失败",
"detail": "\nAuthentication failure : You entered an incorrect user name or password."},
},
}
internal/service/ociconfig.go
+6 -8
View File
@@ -16,11 +16,10 @@ import (
// OciConfigService 管理 API Key 配置的导入、测活与快照同步。
type OciConfigService struct {
db *gorm.DB
cipher *crypto.Cipher
client oci.Client
cleanupTasks *TaskService
cleanupEvents *LogEventService
db *gorm.DB
cipher *crypto.Cipher
client oci.Client
cleanupTasks *TaskService
// auditRaw 按 configId:eventId 暂存审计原始事件(TTL 10 分钟),
// 列表响应剥离 raw 后详情接口据此秒开;miss 走小窗重查兜底。
auditRaw *cache.Cache
@@ -31,10 +30,9 @@ func NewOciConfigService(db *gorm.DB, cipher *crypto.Cipher, client oci.Client)
return &OciConfigService{db: db, cipher: cipher, client: client, auditRaw: cache.New(auditRawMax)}
}
// SetTenantCleanupDeps 注入租户删除提交后的任务与告警内存状态同步依赖。
func (s *OciConfigService) SetTenantCleanupDeps(tasks *TaskService, events *LogEventService) {
// SetTenantCleanupDeps 注入租户删除提交后的任务内存状态同步依赖。
func (s *OciConfigService) SetTenantCleanupDeps(tasks *TaskService) {
s.cleanupTasks = tasks
s.cleanupEvents = events
}
// ImportInput 是导入一份 API Key 的输入：
internal/service/tenantdelete.go
+6 -87
View File
@@ -16,7 +16,6 @@ import (
type tenantDeleteResult struct {
config model.OciConfig
deletedTaskIDs []uint
alertRuleIDs []uint
channelsGone bool
}
@@ -48,7 +47,7 @@ func (s *OciConfigService) deleteTenantInTx(tx *gorm.DB, id uint, result *tenant
if err := s.deleteTenantTasks(tx, id, result); err != nil {
return err
}
if err := deleteTenantEvents(tx, id, result); err != nil {
if err := deleteTenantEvents(tx, id); err != nil {
return err
}
if err := deleteTenantAI(tx, id, result); err != nil {
@@ -205,17 +204,9 @@ func resetTenantTask(tx *gorm.DB, action tenantTaskAction) error {
return nil
}
func deleteTenantEvents(tx *gorm.DB, id uint, result *tenantDeleteResult) error {
ruleIDs, affectedRules, err := loadTenantEventRefs(tx, id)
if err != nil {
return err
}
if err := deleteAlertHits(tx, id); err != nil {
return err
}
result.alertRuleIDs = mergeIDs(ruleIDs, affectedRules)
if err := deleteWhere(tx, &model.AlertRule{}, "oci_config_id = ?", id); err != nil {
return fmt.Errorf("delete tenant alert rules: %w", err)
func deleteTenantEvents(tx *gorm.DB, id uint) error {
if err := lockTenantEventRows(tx, id); err != nil {
return fmt.Errorf("load tenant log events: %w", err)
}
if err := deleteWhere(tx, &model.LogEvent{}, "oci_config_id = ?", id); err != nil {
return fmt.Errorf("delete tenant log events: %w", err)
@@ -223,84 +214,15 @@ func deleteTenantEvents(tx *gorm.DB, id uint, result *tenantDeleteResult) error
return nil
}
func loadTenantEventRefs(tx *gorm.DB, id uint) ([]uint, []uint, error) {
ruleIDs, err := lockedTenantRuleIDs(tx, id)
if err != nil {
return nil, nil, fmt.Errorf("load tenant alert rules: %w", err)
}
if err := lockTenantEventRows(tx, id); err != nil {
return nil, nil, fmt.Errorf("load tenant log events: %w", err)
}
affected, err := alertHitRuleIDs(tx, id)
if err != nil {
return nil, nil, err
}
return ruleIDs, affected, nil
}
func lockedTenantRuleIDs(tx *gorm.DB, id uint) ([]uint, error) {
var rows []model.AlertRule
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").
Where("oci_config_id = ?", id).Order("id").Find(&rows).Error
ids := make([]uint, 0, len(rows))
for _, row := range rows {
ids = append(ids, row.ID)
}
return ids, err
}
// lockTenantEventRows 对租户全部日志事件行加 FOR UPDATE 锁(SQLite 忽略,
// MySQL/PG 阻塞并发命中写入);事件可达数万,ID 不再回传拼接 SQL,
// 后续删除与关联查询一律用子查询,避免绑定变量上限。
// MySQL/PG 阻塞并发写入);事件可达数万,ID 不回传拼接 SQL,删除用条件语句,
// 避免绑定变量上限。
func lockTenantEventRows(tx *gorm.DB, id uint) error {
var rows []model.LogEvent
return tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").
Where("oci_config_id = ?", id).Order("id").Find(&rows).Error
}
// tenantEventIDs 构造「本租户日志事件 ID」子查询,供 IN (?) 内联。
func tenantEventIDs(tx *gorm.DB, id uint) *gorm.DB {
return tx.Model(&model.LogEvent{}).Select("id").Where("oci_config_id = ?", id)
}
// alertHitRuleIDs 找出命中引用了本租户事件的规则 ID(去重,可能含他租户/全局规则)。
func alertHitRuleIDs(tx *gorm.DB, id uint) ([]uint, error) {
ids := make([]uint, 0)
err := tx.Model(&model.AlertRuleHit{}).Where("log_event_id IN (?)", tenantEventIDs(tx, id)).
Distinct().Pluck("rule_id", &ids).Error
if err != nil {
return nil, fmt.Errorf("load affected alert rules: %w", err)
}
return ids, nil
}
func mergeIDs(groups ...[]uint) []uint {
seen := make(map[uint]struct{})
out := make([]uint, 0)
for _, ids := range groups {
for _, id := range ids {
if _, ok := seen[id]; ok {
continue
}
seen[id] = struct{}{}
out = append(out, id)
}
}
return out
}
// deleteAlertHits 删除本租户相关的全部命中:规则属于本租户,或命中引用了
// 本租户的日志事件(他租户/全局规则命中本租户事件的行一并清)。
func deleteAlertHits(tx *gorm.DB, id uint) error {
rules := tx.Model(&model.AlertRule{}).Select("id").Where("oci_config_id = ?", id)
err := tx.Where("rule_id IN (?) OR log_event_id IN (?)", rules, tenantEventIDs(tx, id)).
Delete(&model.AlertRuleHit{}).Error
if err != nil {
return fmt.Errorf("delete tenant alert hits: %w", err)
}
return nil
}
func deleteTenantAI(tx *gorm.DB, id uint, result *tenantDeleteResult) error {
channelIDs, err := lockedAiChannelIDs(tx, id)
if err != nil {
@@ -451,9 +373,6 @@ func (s *OciConfigService) afterTenantDelete(ctx context.Context, result *tenant
if client, ok := s.client.(tenancyCacheInvalidator); ok {
client.InvalidateTenancy(result.config.TenancyOCID)
}
if s.cleanupEvents != nil {
s.cleanupEvents.ClearAlertCooldown(result.alertRuleIDs)
}
if s.cleanupTasks != nil {
s.cleanupTasks.ApplyTenantCleanup(ctx, result.deletedTaskIDs, result.channelsGone)
}
internal/service/tenantdelete_test.go
+4 -65
View File
@@ -55,7 +55,7 @@ func newTenantDeleteEnv(t *testing.T, client oci.Client) (*OciConfigService, *Ta
}
configs := NewOciConfigService(db, cipher, client)
tasks := NewTaskService(db, configs, nil, nil)
configs.SetTenantCleanupDeps(tasks, nil)
configs.SetTenantCleanupDeps(tasks)
return configs, tasks, db
}
@@ -64,7 +64,7 @@ func migrateTenantDeleteModels(t *testing.T, db *gorm.DB) {
err := db.AutoMigrate(
&model.OciConfig{}, &model.Task{}, &model.TaskLog{}, &model.Setting{},
&model.CheckSnapshot{}, &model.CostSnapshot{}, &model.RegionCache{}, &model.CompartmentCache{},
&model.LogEvent{}, &model.AlertRule{}, &model.AlertRuleHit{},
&model.LogEvent{},
&model.AiChannel{}, &model.AiModelCache{}, &model.AiCallLog{}, &model.AiContentLog{},
&model.Proxy{}, &model.AiKey{}, &model.SystemLog{},
)
@@ -162,19 +162,9 @@ func seedTenantEvents(t *testing.T, db *gorm.DB, target, other uint) {
t.Helper()
targetEvent := model.LogEvent{OciConfigID: target, MessageID: "target-event"}
otherEvent := model.LogEvent{OciConfigID: other, MessageID: "other-event"}
targetRule := model.AlertRule{Name: "target-rule", OciConfigID: target}
globalRule := model.AlertRule{Name: "global-rule", OciConfigID: 0}
for _, value := range []any{&targetEvent, &otherEvent, &targetRule, &globalRule} {
for _, value := range []any{&targetEvent, &otherEvent} {
mustCreate(t, db, value)
}
hits := []model.AlertRuleHit{
{RuleID: targetRule.ID, LogEventID: otherEvent.ID},
{RuleID: globalRule.ID, LogEventID: targetEvent.ID},
{RuleID: globalRule.ID, LogEventID: otherEvent.ID},
}
for i := range hits {
mustCreate(t, db, &hits[i])
}
}
func seedTenantAI(t *testing.T, db *gorm.DB, target, other uint) {
@@ -202,7 +192,7 @@ func assertTenantRowsGone(t *testing.T, db *gorm.DB, id uint) {
rows := []any{
&model.OciConfig{}, &model.CheckSnapshot{}, &model.CostSnapshot{},
&model.RegionCache{}, &model.CompartmentCache{}, &model.LogEvent{},
&model.AlertRule{}, &model.AiChannel{},
&model.AiChannel{},
}
for _, value := range rows {
column := "oci_config_id"
@@ -212,7 +202,6 @@ func assertTenantRowsGone(t *testing.T, db *gorm.DB, id uint) {
assertCount(t, db, value, column+" = ?", []any{id}, 0)
}
assertCount(t, db, &model.Setting{}, "key = ?", []any{secretKey(id)}, 0)
assertCount(t, db, &model.AlertRuleHit{}, "", nil, 1)
assertCount(t, db, &model.AiModelCache{}, "", nil, 1)
assertCount(t, db, &model.AiCallLog{}, "", nil, 1)
assertCount(t, db, &model.AiContentLog{}, "", nil, 1)
@@ -248,31 +237,10 @@ func assertRemainingIndirectRows(t *testing.T, db *gorm.DB, otherID uint) {
t.Fatalf("load other AI call: %v", err)
}
assertCount(t, db, &model.AiContentLog{}, "call_log_id = ?", []any{call.ID}, 1)
assertRemainingAlertHit(t, db, otherID)
}
func assertRemainingAlertHit(t *testing.T, db *gorm.DB, otherID uint) {
t.Helper()
var hit model.AlertRuleHit
if err := db.First(&hit).Error; err != nil {
t.Fatalf("load remaining alert hit: %v", err)
}
var rule model.AlertRule
var event model.LogEvent
if err := db.First(&rule, hit.RuleID).Error; err != nil {
t.Fatalf("load remaining rule: %v", err)
}
if err := db.First(&event, hit.LogEventID).Error; err != nil {
t.Fatalf("load remaining event: %v", err)
}
if rule.OciConfigID != 0 || event.OciConfigID != otherID {
t.Errorf("remaining hit = rule cfg %d/event cfg %d, want global/other", rule.OciConfigID, event.OciConfigID)
}
}
func assertRetainedGlobals(t *testing.T, db *gorm.DB) {
t.Helper()
assertCount(t, db, &model.AlertRule{}, "oci_config_id = 0", nil, 1)
assertCount(t, db, &model.Proxy{}, "", nil, 1)
assertCount(t, db, &model.AiKey{}, "", nil, 1)
assertCount(t, db, &model.SystemLog{}, "", nil, 1)
@@ -629,35 +597,6 @@ func TestDeleteTenantSkipsCorruptTaskPayload(t *testing.T) {
assertCount(t, db, &model.OciConfig{}, "id = ?", []any{target.ID}, 0)
}
func TestDeleteTenantEventsAffectedRules(t *testing.T) {
_, _, db := newTenantDeleteEnv(t, &fakeClient{})
target, other := seedDeleteTenants(t, db)
seedTenantEvents(t, db, target.ID, other.ID)
// 规则 ID 须在删除前取:target-rule 会随租户一并删除
var targetRule, globalRule model.AlertRule
db.Where("name = ?", "target-rule").First(&targetRule)
db.Where("name = ?", "global-rule").First(&globalRule)
result := &tenantDeleteResult{}
err := db.Transaction(func(tx *gorm.DB) error {
return deleteTenantEvents(tx, target.ID, result)
})
if err != nil {
t.Fatalf("deleteTenantEvents: %v", err)
}
// target-rule 属本租户,global-rule 的命中引用了本租户事件:都应回收冷却
got := map[uint]bool{}
for _, id := range result.alertRuleIDs {
got[id] = true
}
if len(got) != 2 || !got[targetRule.ID] || !got[globalRule.ID] {
t.Errorf("alertRuleIDs = %v, want 含 target-rule(%d) 与 global-rule(%d)", result.alertRuleIDs, targetRule.ID, globalRule.ID)
}
// 仅保留 global-rule × other-event 一条命中
assertCount(t, db, &model.AlertRuleHit{}, "", nil, 1)
}
func TestDeleteTenantManyEventsNoVarLimit(t *testing.T) {
// 回归:事件数超 SQLite 绑定变量上限(32766)时删除仍成功(旧实现 IN 展开必失败)
configs, _, db := newTenantDeleteEnv(t, &fakeClient{})