Consul #

client.Agent().ServiceRegister(&api.AgentServiceRegistration{
    ID:      "echo-a",
    Name:    "echo-a",
    Address: "echo-a",          // 容器网络内可达的地址（=compose 服务名；Windows 上=127.0.0.1）
    Port:    8080,
    Tags:    buildTags(),       // ★ Traefik 路由规则就藏在 tags 里（见 3.2）
    Check: &api.AgentServiceCheck{
        HTTP:                           "http://echo-a:8080/healthz",
        Interval:                       "5s",
        Timeout:                        "2s",
        DeregisterCriticalServiceAfter: "1m",   // 挂满 1 分钟自动从注册表摘除
    },
})

curl.exe -s http://localhost/a                          # 路由通了（服务自己注册进去的）
curl.exe -s http://localhost:8500/v1/catalog/services   # Consul 里有谁注册了
curl.exe -s http://localhost:8080/api/http/routers | findstr /i consulcatalog
#   "name":"echo-a@consulcatalog" ... "provider":"consulcatalog"  ← 路由来自 Consul，非 docker label

# command 写法（09 Docker）
- "--providers.consulcatalog=true"
- "--providers.consulcatalog.endpoint.address=consul:8500"
- "--providers.consulcatalog.exposedByDefault=false"   # 只暴露带 traefik.enable=true 的
- "--providers.consulcatalog.refreshInterval=5s"        # 5s 拉一次，路由更新更快

# traefik.yml 写法（10 Windows）
providers:
  consulCatalog:
    endpoint:
      address: "127.0.0.1:8500"
      scheme: "http"
    exposedByDefault: false
    refreshInterval: "5s"

# 加路由组：改服务 EXTRA_GROUPS 重启 → 自注册推上去，Traefik ~5s 拿到
curl.exe -s http://localhost/admin-a/   # viaPrefix=/admin-a（新路由组，同一个 echo-a）
# 加接口：只加 handler，无需注册
curl.exe -s http://localhost/a/time         # 经 /a 路由组访问
curl.exe -s http://localhost/admin-a/time   # 经 /admin-a 也能访问（整段前缀早转发进来了）

// 等价于命令行：consul watch -type=keyprefix -prefix=config/
plan, _ := watch.Parse(map[string]interface{}{
    "type":   "keyprefix",
    "prefix": "config/",
})
plan.HybridHandler = func(_ watch.BlockingParamVal, raw interface{}) {
    pairs := raw.(api.KVPairs)         // 首次回调带全量，之后每次变化再回调
    // ... 挑出本服务关心的键(shared/* 与 <name>/*)，原子热替换内存配置
}
go plan.Run(consulAddr)

# 方式1：docker exec consul CLI
docker compose exec -T consul consul kv put config/shared/color green        # 公共，全员变
docker compose exec -T consul consul kv put config/echo-a/greeting "热更新!"  # 私有，只 echo-a 变
# 方式2：Windows 原生
.\bin\consul.exe kv put config/shared/color green
# 方式3：Consul UI → Key/Value 直接编辑保存

# 仅当 key 不存在时写入 → 你热改过的值重跑不会被重置
setdefault() {
  if consul kv get "$1" >/dev/null 2>&1; then echo "skip $1"; else consul kv put "$1" "$2"; fi
}
setdefault config/shared/color "blue"
setdefault config/expected-services '["echo-a","echo-b","monitor","ghost-svc"]'

docker compose kill echo-b           # 模拟崩溃 → 等~10s 面板变红 DOWN
curl.exe -s http://localhost/status  # 对账结果(JSON)
docker compose start echo-b          # 恢复 → ✅ DOWN→HEALTHY

# "谁还没注册"开箱即见：期望表里有永不启动的 ghost-svc → 一直 MISSING
docker compose exec -T consul consul kv get config/expected-services
docker compose exec -T consul consul kv put config/expected-services '["echo-a","echo-b","monitor"]'  # 去掉它，预警消失

cd 09-consul
docker compose up -d --build
docker compose logs -f
docker compose down            # -dev 内存态，down 后配置清空
# 面板 http://localhost/ ｜ Consul http://localhost:8500 ｜ Traefik http://localhost:8080

cd 10-consul-windows
.\download.ps1     # 下载 consul.exe / traefik.exe 到 .\bin（只做一次）
.\build.ps1        # 用 09 源码编译 echo-a/echo-b/monitor.exe（只做一次）
.\start.ps1        # 拉起全部（幂等：已在跑的跳过）→ 面板 http://localhost:8000/
Get-Content logs\monitor.log -Wait    # 看日志（≈ docker logs -f）
.\stop.ps1                            # 全停
.\stop.ps1 echo-b                     # 只停一个

consul.exe agent -server -bootstrap-expect=1 -data-dir .\data `
  -bind=127.0.0.1 -client=127.0.0.1 -ui

docker compose logs -f               # 09：全部服务日志交织
Get-Content logs\monitor.log -Wait   # 10：单服务日志
curl.exe -s http://localhost/status  # 对账+预警 JSON

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "io"
    "log"
    "net/http"
    "net/url"
    "os"
    "os/signal"
    "sort"
    "strconv"
    "strings"
    "sync"
    "syscall"
    "time"

    "github.com/hashicorp/consul/api"
    "github.com/hashicorp/consul/api/watch"
)

func env(k, def string) string {
    if v := os.Getenv(k); v != "" {
       return v
    }
    return def
}

var (
    serviceName = env("SERVICE_NAME", "echo")
    serviceAddr = env("SERVICE_ADDR", "127.0.0.1")
    listenPort  = env("PORT", "8080")
    routePrefix = env("ROUTE_PREFIX", "/")
    extraGroups = env("EXTRA_GROUPS", "")
    consulAddr  = env("CONSUL_ADDR", "http://127.0.0.1:8500")
    kvPrefix    = "config"
)

// ── 当前生效配置（支持热更新） ──
type configState struct {
    mu      sync.RWMutex
    version uint64
    values  map[string]string
}

var cfg = &configState{values: map[string]string{}}

func (c *configState) snapshot() (uint64, map[string]string) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    cp := make(map[string]string, len(c.values))
    for k, v := range c.values {
       cp[k] = v
    }
    return c.version, cp
}

func main() {
    port, _ := strconv.Atoi(listenPort)
    client, addr, err := newConsul()
    if err != nil {
       log.Fatalf("[%s] 创建 Consul 客户端失败: %v", serviceName, err)
    }
    // 1) 注册
    if err := registerWithRetry(client, port); err != nil {
       log.Fatalf("[%s] 注册 Consul 失败: %v", serviceName, err)
    }
    go reRegisterLoop(client, port)
    // 2) 官方 watch 监听 config/ 热更新
    plan, err := startConfigWatch(addr)
    if err != nil {
       log.Fatalf("[%s] 启动配置 watch 失败: %v", serviceName, err)
    }
    // 3) HTTP（路由组都带 stripprefix，所以这里是去掉前缀后的干净路径）
    mux := http.NewServeMux()
    mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "ok") })
    mux.HandleFunc("/time", handleTime)
    mux.HandleFunc("/", handleRoot)
    srv := &http.Server{Addr: ":" + listenPort, Handler: mux}
    go func() {
       log.Printf("[%s] 监听 :%s  路由组=%v  consul=%s", serviceName, listenPort, groupSummary(), consulAddr)
       if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
          log.Fatalf("[%s] http 退出: %v", serviceName, err)
       }
    }()
    // 4) 优雅退出
    stop := make(chan os.Signal, 1)
    signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
    <-stop
    if plan != nil {
       plan.Stop()
    }
    deregister(client)
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    _ = srv.Shutdown(ctx)
}

// ── HTTP handlers ──
func handleRoot(w http.ResponseWriter, r *http.Request) {
    ver, vals := cfg.snapshot()
    writeJSON(w, map[string]any{
       "service": serviceName, "address": serviceAddr, "port": listenPort,
       "viaPrefix": r.Header.Get("X-Forwarded-Prefix"), // 经哪个路由组进来的
       "configVersion": ver, "config": vals, "servedAt": time.Now().Format("15:04:05"),
    })
}

func handleTime(w http.ResponseWriter, r *http.Request) {
    writeJSON(w, map[string]any{
       "service": serviceName, "endpoint": "/time",
       "now": time.Now().Format("2006-01-02 15:04:05"), "viaPrefix": r.Header.Get("X-Forwarded-Prefix"),
       "note": "新接口=只加 handler，无需在 Consul/Traefik 注册",
    })
}

func writeJSON(w http.ResponseWriter, v any) {
    w.Header().Set("Content-Type", "application/json; charset=utf-8")
    _ = json.NewEncoder(w).Encode(v)
}

// ── 路由组 → Traefik tags ──
type routeGroup struct {
    router string // Traefik router 名（全局唯一）
    prefix string // 对外前缀
}

func routeGroups() []routeGroup {
    gs := []routeGroup{{router: serviceName, prefix: routePrefix}}
    for _, item := range strings.Split(extraGroups, ";") {
       item = strings.TrimSpace(item)
       if item == "" {
          continue
       }
       parts := strings.SplitN(item, ":", 2)
       if len(parts) != 2 {
          continue
       }
       gs = append(gs, routeGroup{router: serviceName + "-" + strings.TrimSpace(parts[0]), prefix: strings.TrimSpace(parts[1])})
    }
    return gs
}

func groupSummary() []string {
    var out []string
    for _, g := range routeGroups() {
       out = append(out, g.router+"="+g.prefix)
    }
    return out
}

// buildTags：把每个路由组翻译成 Traefik tags（一条 router + 一个 stripprefix 中间件）
func buildTags() []string {
    tags := []string{"traefik.enable=true"}
    for _, g := range routeGroups() {
       strip := g.router + "-strip"
       tags = append(tags,
          fmt.Sprintf("traefik.http.routers.%s.rule=PathPrefix(`%s`)", g.router, g.prefix),
          fmt.Sprintf("traefik.http.routers.%s.entrypoints=web", g.router),
          fmt.Sprintf("traefik.http.routers.%s.service=%s", g.router, serviceName),
          fmt.Sprintf("traefik.http.routers.%s.middlewares=%s", g.router, strip),
          fmt.Sprintf("traefik.http.middlewares.%s.stripprefix.prefixes=%s", strip, g.prefix),
       )
    }
    return tags
}

// ── 注册 / 注销 ──
func newConsul() (*api.Client, string, error) {
    u, err := url.Parse(consulAddr)
    if err != nil {
       return nil, "", err
    }
    c := api.DefaultConfig()
    c.Address = u.Host
    if u.Scheme != "" {
       c.Scheme = u.Scheme
    }
    cl, err := api.NewClient(c)
    return cl, u.Host, err
}

func register(client *api.Client, port int) error {
    return client.Agent().ServiceRegister(&api.AgentServiceRegistration{
       ID: serviceName, Name: serviceName, Address: serviceAddr, Port: port, Tags: buildTags(),
       Check: &api.AgentServiceCheck{
          HTTP:                           fmt.Sprintf("http://%s:%d/healthz", serviceAddr, port),
          Interval:                       "5s",
          Timeout:                        "2s",
          DeregisterCriticalServiceAfter: "1m",
       },
    })
}

func registerWithRetry(client *api.Client, port int) error {
    var err error
    for i := 0; i < 30; i++ {
       if err = register(client, port); err == nil {
          log.Printf("[%s] ✅ 已注册到 Consul（addr=%s:%d，路由组 %v）", serviceName, serviceAddr, port, groupSummary())
          return nil
       }
       log.Printf("[%s] 注册失败(%d/30): %v，2s 后重试", serviceName, i+1, err)
       time.Sleep(2 * time.Second)
    }
    return err
}

func reRegisterLoop(client *api.Client, port int) {
    t := time.NewTicker(30 * time.Second)
    defer t.Stop()
    for range t.C {
       if err := register(client, port); err != nil {
          log.Printf("[%s] 重注册失败: %v", serviceName, err)
       }
    }
}

func deregister(client *api.Client) {
    if err := client.Agent().ServiceDeregister(serviceName); err != nil {
       log.Printf("[%s] 注销失败: %v", serviceName, err)
       return
    }
    log.Printf("[%s] 已从 Consul 注销", serviceName)
}

// ── 配置热更新（官方 api/watch keyprefix） ──
func startConfigWatch(addr string) (*watch.Plan, error) {
    plan, err := watch.Parse(map[string]interface{}{"type": "keyprefix", "prefix": kvPrefix + "/"})
    if err != nil {
       return nil, err
    }
    plan.HybridHandler = func(_ watch.BlockingParamVal, raw interface{}) {
       if raw == nil {
          return
       }
       pairs, ok := raw.(api.KVPairs)
       if !ok {
          return
       }
       all := map[string]string{}
       var version uint64
       for _, p := range pairs {
          all[p.Key] = string(p.Value)
          if p.ModifyIndex > version {
             version = p.ModifyIndex
          }
       }
       applyConfig(version, all)
    }
    go func() {
       if err := plan.Run(addr); err != nil {
          log.Printf("[%s] 配置 watch 退出: %v", serviceName, err)
       }
    }()
    return plan, nil
}

// applyConfig：从全量 KV 挑出本服务关心的键，原子热替换
func applyConfig(version uint64, all map[string]string) {
    sharedPfx := kvPrefix + "/shared/"
    ownPfx := kvPrefix + "/" + serviceName + "/"
    next := map[string]string{}
    for k, v := range all {
       switch {
       case strings.HasPrefix(k, sharedPfx):
          next["shared/"+strings.TrimPrefix(k, sharedPfx)] = v
       case strings.HasPrefix(k, ownPfx):
          next[strings.TrimPrefix(k, ownPfx)] = v
       }
    }
    cfg.mu.Lock()
    old := cfg.values
    cfg.values = next
    cfg.version = version
    cfg.mu.Unlock()
    if changes := diff(old, next); len(changes) == 0 {
       log.Printf("[%s] 配置加载 version=%d（共 %d 项）", serviceName, version, len(next))
    } else {
       log.Printf("[%s] 🔄 配置热更新 version=%d：%s", serviceName, version, strings.Join(changes, "；"))
    }
}

func diff(a, b map[string]string) []string {
    keys := map[string]bool{}
    for k := range a {
       keys[k] = true
    }
    for k := range b {
       keys[k] = true
    }
    var ks []string
    for k := range keys {
       ks = append(ks, k)
    }
    sort.Strings(ks)
    var out []string
    for _, k := range ks {
       av, ao := a[k]
       bv, bo := b[k]
       switch {
       case !ao && bo:
          out = append(out, fmt.Sprintf("新增 %s=%q", k, bv))
       case ao && !bo:
          out = append(out, fmt.Sprintf("删除 %s", k))
       case av != bv:
          out = append(out, fmt.Sprintf("%s %q→%q", k, av, bv))
       }
    }
    return out
}

FROM golang:1.24-alpine AS build
ENV GOTOOLCHAIN=local
WORKDIR /src
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 go build -trimpath -o /out/echosvc .

FROM alpine:3.20
RUN adduser -D -u 10001 appuser
COPY --from=build /out/echosvc /echosvc
USER appuser
EXPOSE 8080
ENTRYPOINT ["/echosvc"]

package main

import (
    "bytes"
    "context"
    "encoding/base64"
    "encoding/json"
    "fmt"
    "io"
    "log"
    "net/http"
    "os"
    "os/signal"
    "sort"
    "strconv"
    "sync"
    "syscall"
    "time"
)

func env(k, def string) string {
    if v := os.Getenv(k); v != "" {
       return v
    }
    return def
}
func envInt(k string, def int) int {
    if v := os.Getenv(k); v != "" {
       if n, err := strconv.Atoi(v); err == nil {
          return n
       }
    }
    return def
}

var (
    serviceName  = env("SERVICE_NAME", "monitor")
    serviceAddr  = env("SERVICE_ADDR", "127.0.0.1")
    listenPort   = env("PORT", "9000")
    consulAddr   = env("CONSUL_ADDR", "http://127.0.0.1:8500")
    reconcileSec = envInt("RECONCILE_INTERVAL_SEC", 5)
)

var httpClient = &http.Client{Timeout: 10 * time.Second}

type SvcStatus struct {
    Name     string `json:"name"`
    Expected bool   `json:"expected"`
    Status   string `json:"status"` // HEALTHY | DOWN | MISSING | UNEXPECTED
    Healthy  int    `json:"healthy"`
    Total    int    `json:"total"`
    Note     string `json:"note"`
}

type Snapshot struct {
    Time     string            `json:"time"`
    Services []SvcStatus       `json:"services"`
    Config   map[string]string `json:"config"`
    Alerts   []string          `json:"alerts"`
    AnyAlert bool              `json:"anyAlert"`
}

var (
    snapMu     sync.RWMutex
    snapshot   Snapshot
    prevStatus = map[string]string{} // 上一轮各服务状态，用于检测变化预警
    alertsMu   sync.Mutex
    alertsRing []string
)

func setSnapshot(s Snapshot) { snapMu.Lock(); snapshot = s; snapMu.Unlock() }
func getSnapshot() Snapshot  { snapMu.RLock(); defer snapMu.RUnlock(); return snapshot }

func main() {
    port, _ := strconv.Atoi(listenPort)
    if err := registerWithRetry(port); err != nil {
       log.Fatalf("[%s] 注册 Consul 失败: %v", serviceName, err)
    }
    go reRegisterLoop(port)
    go reconcileLoop()
    mux := http.NewServeMux()
    mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "ok") })
    mux.HandleFunc("/status", handleStatus) // ★ 不能用 /api/*，会被 echo-a 的 PathPrefix(`/a`) 抢走
    mux.HandleFunc("/", handleDashboard)
    srv := &http.Server{Addr: ":" + listenPort, Handler: mux}
    go func() {
       log.Printf("[%s] 监听 :%s  对账间隔=%ds  consul=%s", serviceName, listenPort, reconcileSec, consulAddr)
       if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
          log.Fatalf("[%s] http 退出: %v", serviceName, err)
       }
    }()
    stop := make(chan os.Signal, 1)
    signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
    <-stop
    deregister()
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    _ = srv.Shutdown(ctx)
}

func handleStatus(w http.ResponseWriter, r *http.Request) {
    w.Header().Set("Content-Type", "application/json; charset=utf-8")
    _ = json.NewEncoder(w).Encode(getSnapshot())
}
func handleDashboard(w http.ResponseWriter, r *http.Request) {
    w.Header().Set("Content-Type", "text/html; charset=utf-8")
    io.WriteString(w, dashboardHTML) // dashboardHTML 见 dashboard.go（见下方说明）
}

// ── 对账 + 预警 ──
func reconcileLoop() {
    reconcile()
    t := time.NewTicker(time.Duration(reconcileSec) * time.Second)
    defer t.Stop()
    for range t.C {
       reconcile()
    }
}

func reconcile() {
    expected, _ := readExpected()
    catalog, _ := catalogServices()
    cfgMap, _ := kvRecurse("config")
    var statuses []SvcStatus
    seen := map[string]bool{}
    for _, name := range expected {
       seen[name] = true
       st := SvcStatus{Name: name, Expected: true}
       if _, ok := catalog[name]; !ok {
          st.Status = "MISSING"
          st.Note = "在期望表中，但从未注册 / 已掉线被移除"
       } else {
          healthy, total := healthService(name)
          st.Healthy, st.Total = healthy, total
          if healthy > 0 {
             st.Status = "HEALTHY"
          } else {
             st.Status = "DOWN"
             st.Note = "已注册但健康检查未通过"
          }
       }
       statuses = append(statuses, st)
    }
    for name := range catalog {
       if seen[name] || name == "consul" {
          continue
       }
       statuses = append(statuses, SvcStatus{Name: name, Status: "UNEXPECTED", Note: "已注册但不在期望表"})
    }
    sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
    detectAlerts(statuses)
    any := false
    for _, s := range statuses {
       if s.Status == "DOWN" || s.Status == "MISSING" {
          any = true
       }
    }
    setSnapshot(Snapshot{Time: time.Now().Format("15:04:05"), Services: statuses, Config: cfgMap, Alerts: getAlerts(), AnyAlert: any})
}

// detectAlerts 仅在状态变化时告警，避免刷屏
func detectAlerts(statuses []SvcStatus) {
    cur := map[string]string{}
    for _, s := range statuses {
       cur[s.Name] = s.Status
    }
    for name, st := range cur {
       old, ok := prevStatus[name]
       if !ok {
          if st == "DOWN" || st == "MISSING" {
             pushAlert(fmt.Sprintf("🚨 %s 初始即异常：%s", name, st))
          }
          continue
       }
       if old != st {
          icon := "ℹ️"
          switch st {
          case "DOWN", "MISSING":
             icon = "🚨"
          case "HEALTHY":
             icon = "✅"
          }
          pushAlert(fmt.Sprintf("%s %s：%s → %s", icon, name, old, st))
       }
    }
    prevStatus = cur
}

func pushAlert(msg string) {
    line := time.Now().Format("15:04:05") + "  " + msg
    log.Printf("[%s/预警] %s", serviceName, msg)
    alertsMu.Lock()
    alertsRing = append(alertsRing, line)
    if len(alertsRing) > 12 {
       alertsRing = alertsRing[len(alertsRing)-12:]
    }
    alertsMu.Unlock()
}

func getAlerts() []string {
    alertsMu.Lock()
    defer alertsMu.Unlock()
    out := make([]string, len(alertsRing))
    copy(out, alertsRing)
    for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
       out[i], out[j] = out[j], out[i]
    }
    return out
}

// ── Consul 读取助手 ──
func readExpected() ([]string, error) {
    raw, err := kvGetRaw("config/expected-services")
    if err != nil {
       return nil, err
    }
    var arr []string
    if err := json.Unmarshal([]byte(raw), &arr); err != nil {
       return nil, fmt.Errorf("expected-services 不是 JSON 数组: %w", err)
    }
    return arr, nil
}

func kvGetRaw(key string) (string, error) {
    resp, err := httpClient.Get(consulAddr + "/v1/kv/" + key + "?raw")
    if err != nil {
       return "", err
    }
    defer resp.Body.Close()
    if resp.StatusCode != http.StatusOK {
       return "", fmt.Errorf("kv %s status %d", key, resp.StatusCode)
    }
    b, _ := io.ReadAll(resp.Body)
    return string(b), nil
}

func kvRecurse(prefix string) (map[string]string, error) {
    resp, err := httpClient.Get(consulAddr + "/v1/kv/" + prefix + "?recurse=true")
    if err != nil {
       return nil, err
    }
    defer resp.Body.Close()
    if resp.StatusCode == http.StatusNotFound {
       return map[string]string{}, nil
    }
    if resp.StatusCode != http.StatusOK {
       return nil, fmt.Errorf("kv recurse status %d", resp.StatusCode)
    }
    var items []struct {
       Key   string `json:"Key"`
       Value string `json:"Value"`
    }
    if err := json.NewDecoder(resp.Body).Decode(&items); err != nil {
       return nil, err
    }
    out := map[string]string{}
    for _, it := range items {
       dec, _ := base64.StdEncoding.DecodeString(it.Value)
       out[it.Key] = string(dec)
    }
    return out, nil
}

func catalogServices() (map[string][]string, error) {
    resp, err := httpClient.Get(consulAddr + "/v1/catalog/services")
    if err != nil {
       return nil, err
    }
    defer resp.Body.Close()
    if resp.StatusCode != http.StatusOK {
       return nil, fmt.Errorf("catalog status %d", resp.StatusCode)
    }
    var m map[string][]string
    if err := json.NewDecoder(resp.Body).Decode(&m); err != nil {
       return nil, err
    }
    return m, nil
}

// healthService 返回 (健康实例数, 总实例数)；一个实例所有 check 都 passing 才算健康
func healthService(name string) (healthy, total int) {
    resp, err := httpClient.Get(consulAddr + "/v1/health/service/" + name)
    if err != nil {
       return 0, 0
    }
    defer resp.Body.Close()
    if resp.StatusCode != http.StatusOK {
       return 0, 0
    }
    var entries []struct {
       Checks []struct {
          Status string `json:"Status"`
       } `json:"Checks"`
    }
    if err := json.NewDecoder(resp.Body).Decode(&entries); err != nil {
       return 0, 0
    }
    for _, e := range entries {
       total++
       allPassing := true
       for _, c := range e.Checks {
          if c.Status != "passing" {
             allPassing = false
             break
          }
       }
       if allPassing {
          healthy++
       }
    }
    return healthy, total
}

// ── 注册 / 注销（priority=1 让 "/" 兜底） ──
func registrationPayload(port int) map[string]any {
    name := serviceName
    tags := []string{
       "traefik.enable=true",
       fmt.Sprintf("traefik.http.routers.%s.rule=PathPrefix(`/`)", name),
       fmt.Sprintf("traefik.http.routers.%s.entrypoints=web", name),
       fmt.Sprintf("traefik.http.routers.%s.priority=1", name), // 最低优先级 → /a /b 优先
       fmt.Sprintf("traefik.http.routers.%s.service=%s", name, name),
    }
    return map[string]any{
       "ID": name, "Name": name, "Address": serviceAddr, "Port": port, "Tags": tags,
       "Check": map[string]any{
          "HTTP":                           fmt.Sprintf("http://%s:%d/healthz", serviceAddr, port),
          "Interval":                       "5s",
          "Timeout":                        "2s",
          "DeregisterCriticalServiceAfter": "1m",
       },
    }
}

func register(port int) error {
    body, _ := json.Marshal(registrationPayload(port))
    req, _ := http.NewRequest(http.MethodPut, consulAddr+"/v1/agent/service/register", bytes.NewReader(body))
    req.Header.Set("Content-Type", "application/json")
    resp, err := httpClient.Do(req)
    if err != nil {
       return err
    }
    defer resp.Body.Close()
    if resp.StatusCode != http.StatusOK {
       b, _ := io.ReadAll(resp.Body)
       return fmt.Errorf("status %d: %s", resp.StatusCode, string(b))
    }
    return nil
}

func registerWithRetry(port int) error {
    var err error
    for i := 0; i < 30; i++ {
       if err = register(port); err == nil {
          log.Printf("[%s] ✅ 已注册到 Consul（addr=%s:%d）", serviceName, serviceAddr, port)
          return nil
       }
       time.Sleep(2 * time.Second)
    }
    return err
}

func reRegisterLoop(port int) {
    t := time.NewTicker(30 * time.Second)
    defer t.Stop()
    for range t.C {
       _ = register(port)
    }
}

func deregister() {
    req, _ := http.NewRequest(http.MethodPut, consulAddr+"/v1/agent/service/deregister/"+serviceName, nil)
    resp, err := httpClient.Do(req)
    if err != nil {
       return
    }
    resp.Body.Close()
}

name: consul-demo

services:
  consul:
    image: hashicorp/consul:latest
    container_name: cc-consul
    command: agent -dev -client=0.0.0.0 -log-level=info  # -dev 单节点(含UI)，数据在内存
    ports: ["8500:8500"]
    networks: [app]
    healthcheck:
      test: ["CMD", "consul", "members"]
      interval: 5s
      timeout: 3s
      retries: 12

  init:                                    # 一次性灌初始配置后退出
    image: hashicorp/consul:latest
    container_name: cc-init
    depends_on:
      consul:
        condition: service_healthy
    entrypoint: ["/bin/sh", "/seed.sh"]
    volumes: ["./init/seed.sh:/seed.sh:ro"]
    networks: [app]
    restart: "no"

  traefik:                                 # 纯 consulcatalog 发现（无 docker.sock）
    image: traefik:v3.7.1
    container_name: cc-traefik
    command:
      - "--api.insecure=true"
      - "--providers.consulcatalog=true"
      - "--providers.consulcatalog.endpoint.address=consul:8500"
      - "--providers.consulcatalog.exposedByDefault=false"
      - "--providers.consulcatalog.refreshInterval=5s"
      - "--entrypoints.web.address=:80"
      - "--accesslog=true"
      - "--log.level=INFO"
    ports: ["80:80", "8080:8080"]
    depends_on:
      consul:
        condition: service_healthy
    networks: [app]

  echo-a:
    build: ./echosvc
    container_name: cc-echo-a
    environment:
      - SERVICE_NAME=echo-a
      - SERVICE_ADDR=echo-a
      - PORT=8080
      - ROUTE_PREFIX=/a
      - EXTRA_GROUPS=admin:/admin-a
      - CONSUL_ADDR=http://consul:8500
    depends_on:
      consul: {condition: service_healthy}
      init: {condition: service_completed_successfully}
    networks: [app]

  echo-b:
    build: ./echosvc
    container_name: cc-echo-b
    environment:
      - SERVICE_NAME=echo-b
      - SERVICE_ADDR=echo-b
      - PORT=8080
      - ROUTE_PREFIX=/b
      - CONSUL_ADDR=http://consul:8500
    depends_on:
      consul: {condition: service_healthy}
      init: {condition: service_completed_successfully}
    networks: [app]

  monitor:
    build: ./monitor
    container_name: cc-monitor
    environment:
      - SERVICE_NAME=monitor
      - SERVICE_ADDR=monitor
      - PORT=9000
      - CONSUL_ADDR=http://consul:8500
      - RECONCILE_INTERVAL_SEC=5
    depends_on:
      consul: {condition: service_healthy}
      init: {condition: service_completed_successfully}
    networks: [app]

networks:
  app:
    driver: bridge

#!/bin/sh
set -e
export CONSUL_HTTP_ADDR="${CONSUL_HTTP_ADDR:-http://consul:8500}"
setdefault() {
  if consul kv get "$1" >/dev/null 2>&1; then echo "[seed] 跳过(已存在) $1"; else consul kv put "$1" "$2"; fi
}
setdefault config/shared/color  "blue"
setdefault config/shared/banner "Hello from Consul 配置中心"
setdefault config/echo-a/greeting "我是 A 服务"
setdefault config/echo-b/greeting "我是 B 服务"
setdefault config/expected-services '["echo-a","echo-b","monitor","ghost-svc"]'
consul kv get -recurse config/

api:
  insecure: true
  dashboard: true
entryPoints:
  web:
    address: ":8000"    # 避开 80（需管理员/易被占）
providers:
  consulCatalog:
    endpoint:
      address: "127.0.0.1:8500"
      scheme: "http"
    exposedByDefault: false
    refreshInterval: "5s"
accessLog: {}
log:
  level: INFO

$ErrorActionPreference = 'Stop'
[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
$root = $PSScriptRoot; $bin = Join-Path $root 'bin'
New-Item -ItemType Directory -Force -Path $bin | Out-Null

function Get-Tool($name, $url, $exeInZip, $outName) {
  $out = Join-Path $bin $outName
  if (Test-Path $out) { Write-Host "已存在，跳过: $outName"; return }
  $tmp = Join-Path $env:TEMP ('dl-' + [guid]::NewGuid().ToString('N'))
  New-Item -ItemType Directory -Force -Path $tmp | Out-Null
  $zip = Join-Path $tmp "$name.zip"
  Invoke-WebRequest -Uri $url -OutFile $zip
  Expand-Archive -Path $zip -DestinationPath $tmp -Force
  Copy-Item (Join-Path $tmp $exeInZip) $out -Force
  Remove-Item $tmp -Recurse -Force
}
# Consul：checkpoint API 取最新稳定版
$consulVer = (Invoke-RestMethod 'https://checkpoint-api.hashicorp.com/v1/check/consul').current_version
Get-Tool 'consul' "https://releases.hashicorp.com/consul/$consulVer/consul_${consulVer}_windows_amd64.zip" 'consul.exe' 'consul.exe'
# Traefik：GitHub releases/latest
$tag = (Invoke-RestMethod 'https://api.github.com/repos/traefik/traefik/releases/latest' -Headers @{ 'User-Agent' = 'demo' }).tag_name
Get-Tool 'traefik' "https://github.com/traefik/traefik/releases/download/$tag/traefik_${tag}_windows_amd64.zip" 'traefik.exe' 'traefik.exe'

$ErrorActionPreference = 'Stop'
$root = $PSScriptRoot; $bin = Join-Path $root 'bin'
$src  = Join-Path (Split-Path $root -Parent) '09-consul'   # ..\09-consul（复用其源码）
New-Item -ItemType Directory -Force -Path $bin | Out-Null
if (-not (Test-Path (Join-Path $src 'echosvc\main.go'))) { throw "找不到 09-consul 源码：$src" }

Push-Location (Join-Path $src 'echosvc')
& go build -o (Join-Path $bin 'echo-a.exe') .          # echo-a 与 echo-b 同一二进制
if ($LASTEXITCODE -ne 0) { Pop-Location; throw 'go build echosvc 失败' }
Pop-Location
Copy-Item (Join-Path $bin 'echo-a.exe') (Join-Path $bin 'echo-b.exe') -Force

Push-Location (Join-Path $src 'monitor')
& go build -o (Join-Path $bin 'monitor.exe') .
if ($LASTEXITCODE -ne 0) { Pop-Location; throw 'go build monitor 失败' }
Pop-Location

$ErrorActionPreference = 'Stop'
$root = $PSScriptRoot; $bin = Join-Path $root 'bin'; $logs = Join-Path $root 'logs'
New-Item -ItemType Directory -Force -Path $logs | Out-Null
foreach ($f in 'consul.exe','traefik.exe','echo-a.exe','echo-b.exe','monitor.exe') {
  if (-not (Test-Path (Join-Path $bin $f))) { throw "缺少 $f。先 .\download.ps1 再 .\build.ps1" }
}
function Test-Running($name) { [bool](Get-Process -Name $name -ErrorAction SilentlyContinue) }
function Start-Bg($name, $exe, $argList, $envVars) {
  if (Test-Running $name) { Write-Host "已在运行，跳过: $name"; return }
  if ($envVars) { foreach ($k in $envVars.Keys) { Set-Item "env:$k" $envVars[$k] } }
  $out = Join-Path $logs "$name.log"; $err = Join-Path $logs "$name.err.log"
  if ($argList) { Start-Process -FilePath $exe -ArgumentList $argList -NoNewWindow -RedirectStandardOutput $out -RedirectStandardError $err | Out-Null }
  else          { Start-Process -FilePath $exe -NoNewWindow -RedirectStandardOutput $out -RedirectStandardError $err | Out-Null }
}
function Wait-Consul {
  for ($i = 0; $i -lt 40; $i++) {
    try { if (Invoke-RestMethod 'http://127.0.0.1:8500/v1/status/leader' -TimeoutSec 2) { return $true } } catch {}
    Start-Sleep -Milliseconds 800
  }
  return $false
}
$consulAddr = 'http://127.0.0.1:8500'
Start-Bg 'consul' (Join-Path $bin 'consul.exe') @('agent','-dev','-client=127.0.0.1') $null
if (-not (Wait-Consul)) { throw "Consul 未就绪，看 logs\consul.err.log" }
& (Join-Path $root 'seed.ps1')                                  # 灌配置
Start-Bg 'traefik' (Join-Path $bin 'traefik.exe') @("--configFile=$(Join-Path $root 'traefik.yml')") $null
Start-Bg 'echo-a'  (Join-Path $bin 'echo-a.exe')  $null @{ SERVICE_NAME='echo-a'; SERVICE_ADDR='127.0.0.1'; PORT='8081'; ROUTE_PREFIX='/a'; CONSUL_ADDR=$consulAddr }
Start-Bg 'echo-b'  (Join-Path $bin 'echo-b.exe')  $null @{ SERVICE_NAME='echo-b'; SERVICE_ADDR='127.0.0.1'; PORT='8082'; ROUTE_PREFIX='/b'; CONSUL_ADDR=$consulAddr }
Start-Bg 'monitor' (Join-Path $bin 'monitor.exe') $null @{ SERVICE_NAME='monitor'; SERVICE_ADDR='127.0.0.1'; PORT='9000'; CONSUL_ADDR=$consulAddr; RECONCILE_INTERVAL_SEC='5' }

$ErrorActionPreference = 'Stop'
$base = 'http://127.0.0.1:8500/v1/kv/'
try { Invoke-RestMethod -Uri 'http://127.0.0.1:8500/v1/status/leader' -TimeoutSec 2 -ErrorAction Stop | Out-Null }
catch { Write-Host "✗ 连不上 Consul，先 .\start.ps1" -ForegroundColor Yellow; exit 1 }
function Set-Default($k, $v) {
  $exists = $false
  try { Invoke-RestMethod -Uri ($base + $k) -Method Get -ErrorAction Stop | Out-Null; $exists = $true } catch { $exists = $false }
  if ($exists) { Write-Host "  skip $k（已存在）" }
  else { Invoke-RestMethod -Uri ($base + $k) -Method Put -Body ([System.Text.Encoding]::UTF8.GetBytes($v)) | Out-Null; Write-Host "  put  $k" }
}
Set-Default 'config/shared/color'  'blue'
Set-Default 'config/shared/banner' 'Hello from Consul 配置中心'
Set-Default 'config/echo-a/greeting' '我是 A 服务'
Set-Default 'config/echo-b/greeting' '我是 B 服务'
Set-Default 'config/expected-services' '["echo-a","echo-b","monitor","ghost-svc"]'

param([string]$Name)
$ErrorActionPreference = 'SilentlyContinue'
if ($Name) {
  Stop-Process -Name $Name -Force; Write-Host "已停止: $Name"
} else {
  foreach ($n in 'echo-a','echo-b','monitor','traefik','consul') { Stop-Process -Name $n -Force }
  Write-Host "已停止全部。注意：consul 是 -dev 内存态，停掉后 KV 配置会清空。"
}