feat: syncer/exec 物化(GraphQL 落库)+ 两阶段编排 main/server
- plan.go:新 diff 引擎(纯函数);输入=镜像期望状态 BuildFromMirror + LLDAP 实际 LDAP 视图;操作顺序 分组→用户→加组→移组→删除,删除仅 confirmed_leave 且最后阶段; 所有权只认镜像台账,服务账号永不碰;孤儿组只告警不自动删 - exec.go:写走 lldapgql;deleteUser 级联清成员;groupId 用 dept_id→groupId 映射 (物化开始时经 GraphQL Groups 拉取),不凭名字反推 - runner.go:两阶段编排,快照与物化共享单实例互斥锁;物化开始登录一次、 全成功才回写 attrs_hash/groupId/MarkDeleted - main.go:-snapshot/-apply/-dry-run 命令行 + 两套定时任务(快照低频/物化略高频)+ check-config - server.go:镜像视图用户/部门、待确认列表、快照触发、一次性离职确认端点 (不经 Auth,token 即凭证)、skipped 展示 - 安全结构拦截均已配变异验证用例(plan_test):未确认不删、非受管不删、 admin 永不移出、删除最后阶段
This commit is contained in:
+110
-205
@@ -1,58 +1,80 @@
|
||||
package syncer
|
||||
|
||||
// 两阶段编排器:快照(飞书→镜像)与物化(镜像→LLDAP)。
|
||||
//
|
||||
// 这两段共享一把单实例互斥锁,保证快照与物化绝不同时执行(镜像事务化 + 物化
|
||||
// 全成功才回写的前提),否则物化可能读到半新半旧的镜像污染 LDAP(约束 2)。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
"sync"
|
||||
|
||||
"idsync/internal/config"
|
||||
"idsync/internal/feishu"
|
||||
"idsync/internal/ldaplib"
|
||||
"idsync/internal/lldapgql"
|
||||
"idsync/internal/mapping"
|
||||
"idsync/internal/store"
|
||||
)
|
||||
|
||||
// Runner 编排一次完整的同步:读取飞书 → 读取 LDAP → 计算计划 → (可选)执行 → 回写台账。
|
||||
// Runner 编排快照与物化。
|
||||
type Runner struct {
|
||||
cfg *config.Config
|
||||
feishu *feishu.Client
|
||||
ldap *ldaplib.Client
|
||||
store *store.Store
|
||||
log *slog.Logger
|
||||
cfg *config.Config
|
||||
feishu *feishu.Client
|
||||
ldap *ldaplib.Client
|
||||
gql *lldapgql.Client
|
||||
store *store.Store
|
||||
snapshot *SnapshotRunner
|
||||
log *slog.Logger
|
||||
|
||||
// guard 是执行前的安全阈值。
|
||||
guard Guard
|
||||
// mu 让快照与物化在单实例内互斥。
|
||||
mu sync.Mutex
|
||||
groupIDs map[string]int
|
||||
}
|
||||
|
||||
// NewRunner 构造编排器。
|
||||
func NewRunner(cfg *config.Config, f *feishu.Client, l *ldaplib.Client, s *store.Store, log *slog.Logger) *Runner {
|
||||
func NewRunner(cfg *config.Config, f *feishu.Client, l *ldaplib.Client, g *lldapgql.Client,
|
||||
s *store.Store, log *slog.Logger, opts SnapshotOptions) *Runner {
|
||||
return &Runner{
|
||||
cfg: cfg,
|
||||
feishu: f,
|
||||
ldap: l,
|
||||
store: s,
|
||||
log: log,
|
||||
cfg: cfg,
|
||||
feishu: f,
|
||||
ldap: l,
|
||||
gql: g,
|
||||
store: s,
|
||||
snapshot: NewSnapshotRunner(cfg, f, s, log, opts),
|
||||
log: log,
|
||||
guard: Guard{
|
||||
MaxDeactivate: cfg.Sync.MaxDeactivate,
|
||||
MaxDeactivatePercent: cfg.Sync.MaxDeactivatePercent,
|
||||
MaxDeactivate: cfg.Guard.MaxDeactivate,
|
||||
MaxDeactivatePercent: cfg.Guard.MaxDeactivatePercent,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// Preview 只计算计划,不做任何写入。dry-run 与实际执行走同一份计划代码,
|
||||
// 因此预览内容与真正会发生的事情一致。
|
||||
// RunSnapshot 触发一轮快照(飞书 → 镜像 + 离职流程推进)。
|
||||
func (r *Runner) RunSnapshot(ctx context.Context) (*SnapshotRunResult, error) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.snapshot.RunSnapshot(ctx)
|
||||
}
|
||||
|
||||
// Preview 只计算物化计划,不做任何写入。dry-run 与真正执行共用同一份计划代码。
|
||||
func (r *Runner) Preview(ctx context.Context) (*Plan, error) {
|
||||
plan, _, err := r.plan(ctx)
|
||||
return plan, err
|
||||
}
|
||||
|
||||
// Run 执行一次同步。dryRun 为 true 时只计算并返回计划。
|
||||
// Run 执行一轮物化(镜像 → LLDAP)。dryRun 为 true 时只计算并返回计划。
|
||||
func (r *Runner) Run(ctx context.Context, dryRun bool, force bool, trigger string) (*Plan, *Result, error) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
|
||||
run, err := r.store.StartRun(modeOf(dryRun), trigger)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("创建同步记录失败: %w", err)
|
||||
return nil, nil, fmt.Errorf("创建物化记录失败: %w", err)
|
||||
}
|
||||
|
||||
plan, desired, err := r.prepare(ctx, run, force)
|
||||
@@ -63,7 +85,7 @@ func (r *Runner) Run(ctx context.Context, dryRun bool, force bool, trigger strin
|
||||
fillRunFromPlan(run, plan)
|
||||
run.Status = "success"
|
||||
if err := r.store.FinishRun(run); err != nil {
|
||||
r.log.Warn("写入同步记录失败", "err", err)
|
||||
r.log.Warn("写入物化记录失败", "err", err)
|
||||
}
|
||||
return plan, nil, nil
|
||||
}
|
||||
@@ -71,17 +93,16 @@ func (r *Runner) Run(ctx context.Context, dryRun bool, force bool, trigger strin
|
||||
res := r.apply(ctx, plan)
|
||||
fillRunFromResult(run, plan, res)
|
||||
|
||||
// 只有全部成功才回写台账:部分失败时保持台账不变,
|
||||
// 让下一轮重新计算差异,避免把未真正落地的变更记成“已完成”。
|
||||
// 只有全部成功才回写镜像:部分失败时保持镜像不变,下轮重算差异。
|
||||
if res.Failed == 0 {
|
||||
if err := r.persist(plan, desired); err != nil {
|
||||
if err := r.persist(ctx, plan, desired); err != nil {
|
||||
run.Status = "failed"
|
||||
run.ErrorMsg = "回写台账失败: " + err.Error()
|
||||
run.ErrorMsg = "回写镜像失败: " + err.Error()
|
||||
_ = r.store.FinishRun(run)
|
||||
return plan, res, fmt.Errorf("回写台账失败: %w", err)
|
||||
return plan, res, fmt.Errorf("回写镜像失败: %w", err)
|
||||
}
|
||||
} else {
|
||||
r.log.Warn("存在失败的操作,跳过台账回写", "failed", res.Failed)
|
||||
r.log.Warn("存在失败的操作,跳过镜像回写", "failed", res.Failed)
|
||||
}
|
||||
|
||||
run.Status = "success"
|
||||
@@ -90,14 +111,14 @@ func (r *Runner) Run(ctx context.Context, dryRun bool, force bool, trigger strin
|
||||
run.ErrorMsg = strings.Join(res.Errors, "; ")
|
||||
}
|
||||
if err := r.store.FinishRun(run); err != nil {
|
||||
r.log.Warn("写入同步记录失败", "err", err)
|
||||
r.log.Warn("写入物化记录失败", "err", err)
|
||||
}
|
||||
|
||||
return plan, res, nil
|
||||
}
|
||||
|
||||
// prepare 计算计划并做安全检查。
|
||||
func (r *Runner) prepare(ctx context.Context, run *store.SyncRun, force bool) (*Plan, *mapping.State, error) {
|
||||
func (r *Runner) prepare(ctx context.Context, run *store.SyncRun, force bool) (*Plan, *mapping.DesiredState, error) {
|
||||
plan, desired, err := r.plan(ctx)
|
||||
if err != nil {
|
||||
run.Status = "failed"
|
||||
@@ -106,8 +127,6 @@ func (r *Runner) prepare(ctx context.Context, run *store.SyncRun, force bool) (*
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
// 阈值检查对 dry-run 也执行,让用户在预览阶段就知道本次变更规模异常。
|
||||
// 只有显式传入 force 才跳过,避免把“预览”与“放行”混为一谈。
|
||||
guard := r.guard
|
||||
if force {
|
||||
guard.AllowBulk = true
|
||||
@@ -121,215 +140,104 @@ func (r *Runner) prepare(ctx context.Context, run *store.SyncRun, force bool) (*
|
||||
return plan, desired, nil
|
||||
}
|
||||
|
||||
// plan 读取两侧状态并计算计划,同时把计划规模记录到 run。
|
||||
func (r *Runner) plan(ctx context.Context) (*Plan, *mapping.State, error) {
|
||||
snap, err := r.feishu.Fetch(ctx, r.cfg.Feishu.DeptList)
|
||||
// plan 读两侧状态并计算物化计划。
|
||||
func (r *Runner) plan(ctx context.Context) (*Plan, *mapping.DesiredState, error) {
|
||||
// 期望侧:只读 SQLite 镜像的 active 视图,不碰飞书。
|
||||
activeUsers, err := r.store.ActiveUsers()
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("读取飞书数据失败: %w", err)
|
||||
return nil, nil, err
|
||||
}
|
||||
r.log.Info("已读取飞书数据",
|
||||
"departments", len(snap.Departments), "users", len(snap.Users),
|
||||
"skipped_no_email", len(snap.SkippedNoEmail), "duplicates", len(snap.SkippedDuplicate))
|
||||
|
||||
// 把“被跳过的用户”显式转成告警:这些人不会出现在 LDAP 里,
|
||||
// 如果不提示,管理者只能靠人肉比对才能发现有人没同步进来。
|
||||
var skipWarnings []string
|
||||
for _, u := range snap.SkippedNoEmail {
|
||||
skipWarnings = append(skipWarnings, fmt.Sprintf(
|
||||
"用户 %s(union_id=%s)没有企业邮箱,无法推导登录名,已跳过",
|
||||
feishu.DisplayName(u), u.UnionID))
|
||||
}
|
||||
for _, d := range snap.SkippedDuplicate {
|
||||
skipWarnings = append(skipWarnings, fmt.Sprintf(
|
||||
"企业邮箱前缀 %s 被多个用户占用(%s),为安全起见全部跳过,请人工确认后修正邮箱",
|
||||
d.UID, strings.Join(d.Names, "、")))
|
||||
depts, err := r.store.ActiveDepartments()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
desired := mapping.BuildFromMirror(activeUsers, depts, r.cfg.Ldap.BaseDN)
|
||||
|
||||
desired := mapping.Build(snap, mapping.BuildOptions{
|
||||
UserOUDN: r.cfg.UserOUDN(),
|
||||
InactiveOUDN: r.cfg.InactiveOUDN(),
|
||||
GroupRootDN: r.cfg.GroupRootDN(),
|
||||
})
|
||||
|
||||
// 实际侧:读 LLDAP 的 LDAP 视图。
|
||||
conn, err := r.ldap.Dial()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
active, err := conn.ListUsers(r.cfg.UserOUDN())
|
||||
actualUsers, err := conn.ListUsers(r.cfg.UserOUDN())
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
inactive, err := conn.ListUsers(r.cfg.InactiveOUDN())
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
groups, err := conn.ListGroups(r.cfg.GroupRootDN())
|
||||
actualGroups, err := conn.ListGroups(r.cfg.GroupRootDN())
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
accounts, err := r.store.AccountsByUID()
|
||||
// 所有权与删除白名单都来自镜像台账。
|
||||
all, err := r.store.AllMirrorUsers()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
groupLedger, err := r.store.GroupsByDeptID()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
ledgerAccounts := make(map[string]string, len(accounts))
|
||||
for uid, a := range accounts {
|
||||
ledgerAccounts[uid] = a.State
|
||||
}
|
||||
ledgerGroups := make(map[string]string, len(groupLedger))
|
||||
for deptID, g := range groupLedger {
|
||||
ledgerGroups[deptID] = g.DN
|
||||
owned := make(map[string]struct{}, len(all))
|
||||
confirmed := make(map[string]bool)
|
||||
for _, u := range all {
|
||||
if u.Skipped {
|
||||
continue
|
||||
}
|
||||
owned[u.UID] = struct{}{}
|
||||
if u.State == store.StateConfirmedLeave {
|
||||
confirmed[u.UID] = true
|
||||
}
|
||||
}
|
||||
|
||||
plan := ComputePlan(&Input{
|
||||
Desired: desired,
|
||||
Active: active,
|
||||
Inactive: inactive,
|
||||
Groups: groups,
|
||||
LedgerAccounts: ledgerAccounts,
|
||||
LedgerGroups: ledgerGroups,
|
||||
ActualUsers: actualUsers,
|
||||
ActualGroups: actualGroups,
|
||||
OwnedUsers: owned,
|
||||
ConfirmedLeave: confirmed,
|
||||
AdminDN: r.cfg.Ldap.AdminDN,
|
||||
PeopleOU: r.cfg.UserOUDN(),
|
||||
GroupRootDN: r.cfg.GroupRootDN(),
|
||||
UserOUDN: r.cfg.UserOUDN(),
|
||||
InactiveOUDN: r.cfg.InactiveOUDN(),
|
||||
})
|
||||
|
||||
plan.Warnings = append(skipWarnings, plan.Warnings...)
|
||||
|
||||
// 对“候选停用”的用户,向飞书确认是否真的离职。
|
||||
// 缺席也可能由权限范围收窄或部门调整造成,确认后再停用可避免误伤。
|
||||
if err := r.confirmResigned(ctx, plan, snap); err != nil {
|
||||
r.log.Warn("确认离职状态时出错,相关账号本轮不停用", "err", err)
|
||||
}
|
||||
|
||||
return plan, desired, nil
|
||||
}
|
||||
|
||||
// confirmResigned 把未能确认离职的停用操作从计划中剔除,并追加告警。
|
||||
func (r *Runner) confirmResigned(ctx context.Context, plan *Plan, _ *feishu.Snapshot) error {
|
||||
var uids []string
|
||||
|
||||
// 只有台账里记录了 union_id 的账号才能精确回查;缺失者无从确认。
|
||||
accounts, err := r.store.AccountsByUID()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for i := range plan.Ops {
|
||||
if plan.Ops[i].Kind != OpDeactivate {
|
||||
continue
|
||||
}
|
||||
acc, ok := accounts[plan.Ops[i].Target]
|
||||
if !ok || acc.UnionID == "" {
|
||||
continue
|
||||
}
|
||||
uids = append(uids, acc.UnionID)
|
||||
}
|
||||
|
||||
if len(uids) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
resigned, err := r.feishu.CheckResigned(ctx, uids)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
confirmed := make(map[string]bool, len(resigned))
|
||||
for _, rr := range resigned {
|
||||
confirmed[rr.UnionID] = true
|
||||
}
|
||||
|
||||
keep := make([]Op, 0, len(plan.Ops))
|
||||
for _, op := range plan.Ops {
|
||||
if op.Kind != OpDeactivate {
|
||||
keep = append(keep, op)
|
||||
continue
|
||||
}
|
||||
acc := accounts[op.Target]
|
||||
if acc != nil && confirmed[acc.UnionID] {
|
||||
keep = append(keep, op)
|
||||
continue
|
||||
}
|
||||
// 未能确认离职:不动作,并明确告知。
|
||||
plan.Stats.Deactivate--
|
||||
plan.Warnings = append(plan.Warnings, fmt.Sprintf(
|
||||
"账号 %s 在飞书已不可见,但未能确认离职状态(可能因通讯录权限范围变化),本轮不迁入 %s",
|
||||
op.Target, r.cfg.InactiveOUDN()))
|
||||
}
|
||||
plan.Ops = keep
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// apply 执行计划。
|
||||
// apply 通过 GraphQL 执行计划。每次物化开始时登录一次,并拉全量组建立 dept_id→groupId 映射。
|
||||
func (r *Runner) apply(ctx context.Context, plan *Plan) *Result {
|
||||
exec := NewExecutor(r.ldap, ExecConfig{
|
||||
UserOUDN: r.cfg.UserOUDN(),
|
||||
InactiveOUDN: r.cfg.InactiveOUDN(),
|
||||
GroupRootDN: r.cfg.GroupRootDN(),
|
||||
GroupRootName: r.cfg.Ldap.GroupRootName,
|
||||
AdminDN: r.cfg.Ldap.AdminDN,
|
||||
InitialPassword: r.cfg.Ldap.InitialPassword,
|
||||
}, r.log)
|
||||
return exec.Apply(ctx, plan)
|
||||
if err := r.gql.Login(ctx, r.cfg.Lldap.AdminUser, r.cfg.Lldap.AdminPass); err != nil {
|
||||
return &Result{Failed: 1, Errors: []string{err.Error()}}
|
||||
}
|
||||
groups, err := r.gql.Groups(ctx)
|
||||
if err != nil {
|
||||
return &Result{Failed: 1, Errors: []string{err.Error()}}
|
||||
}
|
||||
groupIDs := make(map[string]int, len(groups))
|
||||
for _, g := range groups {
|
||||
groupIDs[g.DisplayName] = g.ID
|
||||
}
|
||||
|
||||
exec := NewExecutor(r.gql, ExecConfig{PeopleOU: r.cfg.UserOUDN()}, r.log)
|
||||
res := exec.Apply(ctx, plan, groupIDs)
|
||||
r.groupIDs = groupIDs
|
||||
return res
|
||||
}
|
||||
|
||||
// persist 把本次同步的期望状态回写台账。
|
||||
//
|
||||
// 关键点:按**期望状态**整体刷新,而不是只记录本次实际执行的操作。
|
||||
// 否则会出现一个隐蔽问题——LDAP 中已存在、但台账里没有的分组永远不会被登记
|
||||
// (因为它不需要新建,也就不会产生操作),于是本工具永远不认为它归自己管。
|
||||
// 这个场景在接管既有目录时是常态,因此必须按期望状态登记。
|
||||
func (r *Runner) persist(plan *Plan, desired *mapping.State) error {
|
||||
now := time.Now()
|
||||
|
||||
// 分组:凡在本次期望范围内的都登记。
|
||||
for _, g := range desired.Groups {
|
||||
if err := r.store.UpsertGroup(&store.Group{
|
||||
DeptID: g.DeptID,
|
||||
DN: g.DN,
|
||||
Name: g.Name,
|
||||
ParentID: g.ParentID,
|
||||
State: "active",
|
||||
LastSyncedAt: now,
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// 用户:在职的登记为 active。
|
||||
// persist 在物化全部成功后回写镜像:用户 attrs_hash、部门 groupId、被删用户标记 deleted。
|
||||
func (r *Runner) persist(ctx context.Context, plan *Plan, desired *mapping.DesiredState) error {
|
||||
for _, u := range desired.Users {
|
||||
if err := r.store.UpsertAccount(&store.Account{
|
||||
UID: u.UID,
|
||||
DN: u.DN,
|
||||
Email: u.Mail,
|
||||
Name: u.DisplayName,
|
||||
State: "active",
|
||||
AttrsHash: AttrsHash(u),
|
||||
DeptIDs: strings.Join(u.DeptIDs, ","),
|
||||
LastSyncedAt: now,
|
||||
}); err != nil {
|
||||
if err := r.store.SetUserAttrsHash(u.UID, AttrsHash(u)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
for deptID, gid := range r.groupIDs {
|
||||
if err := r.store.SetDeptGroupID(deptID, gid); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// 停用的账号:更新状态与,DN 指向离职 OU。
|
||||
for _, op := range plan.Ops {
|
||||
if op.Kind == OpDeactivate {
|
||||
if err := r.store.SetAccountState(op.Target, "inactive", op.DN); err != nil {
|
||||
if op.Kind == OpDeleteUser {
|
||||
if err := r.store.MarkDeleted(op.Target); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -341,24 +249,21 @@ func modeOf(dryRun bool) string {
|
||||
return "apply"
|
||||
}
|
||||
|
||||
// fillRunFromPlan 用计划统计填充同步记录(dry-run 时使用)。
|
||||
// fillRunFromPlan 用计划统计填充物化记录(dry-run 时使用)。
|
||||
func fillRunFromPlan(run *store.SyncRun, plan *Plan) {
|
||||
run.Created = plan.Stats.CreateUser + plan.Stats.CreateGroup
|
||||
run.Updated = plan.Stats.UpdateUser + plan.Stats.UpdateGroup +
|
||||
plan.Stats.Reactivate + plan.Stats.Deactivate
|
||||
run.Moved = plan.Stats.Deactivate + plan.Stats.Reactivate
|
||||
run.Updated = plan.Stats.UpdateUser
|
||||
run.Added = plan.Stats.AddMember
|
||||
run.Removed = plan.Stats.RemoveMember
|
||||
run.Skipped = plan.Stats.Unchanged
|
||||
}
|
||||
|
||||
// fillRunFromResult 用实际执行结果填充同步记录。
|
||||
// fillRunFromResult 用实际执行结果填充物化记录。
|
||||
func fillRunFromResult(run *store.SyncRun, plan *Plan, res *Result) {
|
||||
run.Created = res.Created
|
||||
run.Updated = res.Updated
|
||||
run.Moved = res.Moved
|
||||
run.Added = res.Added
|
||||
run.Removed = res.Removed
|
||||
run.Failed = res.Failed
|
||||
run.Skipped = plan.Stats.Unchanged
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user