1
// control_plane.go — 控制面：配置 API + 配置存储
2
package main
3

4
import (
5
  "encoding/json"
6
  "fmt"
7
  "net/http"
8
  "sync"
9
)
10

11
// RouteRule 路由规则：匹配路径前缀，转发到目标地址
12
type RouteRule struct {
13
  Prefix   string `json:"prefix"`   // 路径前缀，如 "/api"
14
  Backend  string `json:"backend"`  // 后端地址，如 "http://localhost:8081"
15
  Priority int    `json:"priority"` // 优先级，数字越大越优先
16
}
17

18
// ConfigStore 配置存储，支持原子替换
19
type ConfigStore struct {
20
  mu     sync.RWMutex
21
  rules  []RouteRule
22
  // version 用于让数据面感知配置是否变更
23
  version int64
24
}
25

26
// GetRules 读取当前路由规则（数据面调用，无锁读）
27
func (s *ConfigStore) GetRules() ([]RouteRule, int64) {
28
  s.mu.RLock()
29
  defer s.mu.RUnlock()
30
  // 返回副本，避免数据面持有锁期间配置被修改
31
  rules := make([]RouteRule, len(s.rules))
32
  copy(rules, s.rules)
33
  return rules, s.version
34
}
35

36
// UpdateRules 更新路由规则（控制面调用）
37
func (s *ConfigStore) UpdateRules(rules []RouteRule) {
38
  s.mu.Lock()
39
  defer s.mu.Unlock()
40
  s.rules = rules
41
  s.version++
42
}
43

44
// ControlPlane 控制面 HTTP 服务器
45
type ControlPlane struct {
46
  store *ConfigStore
47
}
48

49
func (cp *ControlPlane) HandleGetRules(w http.ResponseWriter, r *http.Request) {
50
  rules, version := cp.store.GetRules()
51
  w.Header().Set("X-Config-Version", fmt.Sprintf("%d", version))
52
  json.NewEncoder(w).Encode(rules)
53
}
54

55
func (cp *ControlPlane) HandleUpdateRules(w http.ResponseWriter, r *http.Request) {
56
  var rules []RouteRule
57
  if err := json.NewDecoder(r.Body).Decode(&rules); err != nil {
58
    http.Error(w, err.Error(), http.StatusBadRequest)
59
    return
60
  }
61
  cp.store.UpdateRules(rules)
62
  w.WriteHeader(http.StatusNoContent)
63
}

1
// data_plane.go — 数据面：反向代理，按路由规则转发请求
2
package main
3

4
import (
5
  "net/http"
6
  "net/http/httputil"
7
  "net/url"
8
  "strings"
9
  "sync/atomic"
10
)
11

12
// DataPlane 数据面：从 ConfigStore 读取规则，转发请求
13
type DataPlane struct {
14
  store       *ConfigStore
15
  // 缓存当前配置快照，避免每次请求都读 store
16
  cachedRules atomic.Value // 存储 []RouteRule
17
  cachedVer   atomic.Int64
18
}
19

20
// matchRule 匹配路由规则，返回后端地址
21
func (dp *DataPlane) matchRule(path string) (string, bool) {
22
  // 检查缓存是否最新：对比版本号
23
  rules, ver := dp.store.GetRules()
24
  if ver != dp.cachedVer.Load() {
25
    dp.cachedRules.Store(rules)
26
    dp.cachedVer.Store(ver)
27
  }
28

29
  rules := dp.cachedRules.Load().([]RouteRule)
30
  // 按优先级降序匹配（优先级高的先匹配）
31
  for _, rule := range rules {
32
    if strings.HasPrefix(path, rule.Prefix) {
33
      return rule.Backend, true
34
    }
35
  }
36
  return "", false
37
}
38

39
// ServeHTTP 处理每个请求——这是热路径
40
func (dp *DataPlane) ServeHTTP(w http.ResponseWriter, r *http.Request) {
41
  backend, found := dp.matchRule(r.URL.Path)
42
  if !found {
43
    http.Error(w, "no matching route", http.StatusNotFound)
44
    return
45
  }
46

47
  // 反向代理到后端
48
  target, _ := url.Parse(backend)
49
  proxy := httputil.NewSingleHostReverseProxy(target)
50
  proxy.ServeHTTP(w, r)
51
}

1
// config-server.ts — 控制面：配置管理服务
2
interface RouteRule {
3
  prefix: string;
4
  backend: string;
5
  priority: number;
6
}
7

8
// 配置存储，支持版本号追踪
9
class ConfigStore {
10
  private rules: RouteRule[] = [];
11
  private version = 0;
12

13
  getRules(): { rules: RouteRule[]; version: number } {
14
    // 返回快照，避免调用方持有引用期间配置被修改
15
    return {
16
      rules: [...this.rules],
17
      version: this.version,
18
    };
19
  }
20

21
  updateRules(rules: RouteRule[]): number {
22
    this.rules = rules;
23
    return ++this.version;
24
  }
25
}
26

27
// 控制面 HTTP 服务
28
import express from "express";
29

30
const store = new ConfigStore();
31
const app = express();
32
app.use(express.json());
33

34
// 获取当前路由规则
35
app.get("/api/rules", (_req, res) => {
36
  const { rules, version } = store.getRules();
37
  res.setHeader("X-Config-Version", String(version));
38
  res.json(rules);
39
});
40

41
// 更新路由规则
42
app.put("/api/rules", (req, res) => {
43
  const version = store.updateRules(req.body);
44
  res.setHeader("X-Config-Version", String(version));
45
  res.status(204).end();
46
});
47

48
app.listen(9000, () => console.log("控制面监听 :9000"));

1
// request-handler.ts — 数据面：请求处理器，定期拉取配置
2
import http from "http";
3

4
interface RouteRule {
5
  prefix: string;
6
  backend: string;
7
  priority: number;
8
}
9

10
// 配置缓存，定期从控制面拉取
11
class ConfigCache {
12
  private rules: RouteRule[] = [];
13
  private version = -1;
14
  private configUrl: string;
15

16
  constructor(configUrl: string, private intervalMs = 5000) {
17
    this.configUrl = configUrl;
18
    this.startPull();
19
  }
20

21
  // 定期拉取配置（Pull 模型）
22
  private startPull() {
23
    setInterval(async () => {
24
      try {
25
        const resp = await fetch(this.configUrl);
26
        const serverVersion = Number(resp.headers.get("X-Config-Version"));
27
        // 版本号没变，跳过解析
28
        if (serverVersion === this.version) return;
29
        this.rules = await resp.json();
30
        this.version = serverVersion;
31
      } catch {
32
        // 拉取失败，继续用旧配置——数据面不能因为控制面不可用而停止
33
        console.error("配置拉取失败，继续使用旧配置");
34
      }
35
    }, this.intervalMs);
36
  }
37

38
  // 匹配路由规则（热路径，纯同步操作）
39
  match(path: string): string | null {
40
    for (const rule of this.rules) {
41
      if (path.startsWith(rule.prefix)) {
42
        return rule.backend;
43
      }
44
    }
45
    return null;
46
  }
47
}
48

49
// 数据面 HTTP 服务
50
const cache = new ConfigCache("http://localhost:9000/api/rules");
51

52
const server = http.createServer((req, res) => {
53
  const backend = cache.match(req.url ?? "/");
54
  if (!backend) {
55
    res.writeHead(404);
56
    res.end("no matching route");
57
    return;
58
  }
59

60
  // 简单反向代理：转发请求到后端
61
  const proxyReq = http.request(backend + req.url, {
62
    method: req.method,
63
    headers: req.headers,
64
  }, (proxyRes) => {
65
    res.writeHead(proxyRes.statusCode ?? 502, proxyRes.headers);
66
    proxyRes.pipe(res);
67
  });
68
  proxyReq.on("error", () => {
69
    res.writeHead(502);
70
    res.end("backend unavailable");
71
  });
72
  req.pipe(proxyReq);
73
});
74

75
server.listen(8080, () => console.log("数据面监听 :8080"));