hpnn-proto/hc/neigh.go

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/exec"
	"time"
)

// Резолв MAC членов пула через neigh-таблицу ядра — шаг 5.
//
// Адреса членов известны из конфигурации заранее, их MAC — нет (в реальном
// окружении сервер ещё не подключён, NIC может замениться). hcif-порты —
// настоящие L3-интерфейсы ядра в том же сегменте, что и члены пула, поэтому
// ядро само резолвит их MAC обычным ARP (entrypoint.sh больше не подсовывает
// статические permanent-записи). hcd лишь читает то, что уже узнало ядро, и
// отражает это в единственном месте датапаса, где применяется MAC следующего
// перехода, — в таблице 21 (см. pipeline.sh).
//
// Для клиентов (адрес заранее не известен и не enumerable) резолв остаётся
// пассивным действием learn в самом OpenFlow — здесь не участвует.

// neighEntry — одна запись `ip -json neigh show`.
type neighEntry struct {
	Dst    string   `json:"dst"`
	Dev    string   `json:"dev"`
	Lladdr string   `json:"lladdr"`
	State  []string `json:"state"`
}

// resolvedStates — состояния ARP, при которых MAC считается годным к
// использованию. FAILED и INCOMPLETE (или отсутствие записи) означают, что
// адрес пока не резолвлен.
var resolvedStates = map[string]bool{
	"REACHABLE": true,
	"STALE":     true,
	"DELAY":     true,
	"PROBE":     true,
	"PERMANENT": true,
}

// resolveMAC возвращает MAC адреса ip на интерфейсе iface, если ядро его уже
// резолвило. Пустая строка — адрес пока не резолвлен либо недостижим.
func resolveMAC(iface, ip string) (string, error) {
	out, err := exec.Command("ip", "-json", "neigh", "show", "dev", iface, "to", ip).Output()
	if err != nil {
		return "", err
	}
	var entries []neighEntry
	if err := json.Unmarshal(out, &entries); err != nil {
		return "", fmt.Errorf("разбор ip -json neigh: %w", err)
	}
	for _, e := range entries {
		if e.Lladdr == "" {
			continue
		}
		for _, s := range e.State {
			if resolvedStates[s] {
				return e.Lladdr, nil
			}
		}
	}
	return "", nil
}

// syncNeighbors проходит по всем членам, резолвит их MAC через ядро и для
// каждого изменения атомарно обновляет его правило в таблице adjacency —
// точечным бандлом, не трогая записи остальных членов. Вызывается на том же
// тикере, что и health-пробы.
//
// force заливает правило заново, даже если MAC не изменился с прошлого раза:
// нужно после apply.sh (replace-flows стирает весь пайплайн, включая уже
// залитые сюда правила, а resolveMAC() в следующий раз вернёт тот же MAC и
// без force заливка не повторится). Тот же приём, что у Agent.apply(force).
func (a *Agent) syncNeighbors(force bool) {
	for _, m := range a.cfg.Members {
		mac, err := resolveMAC(m.Iface, m.Address)
		if err != nil {
			log.Printf("neigh: %s (%s@%s): %v", m.Name, m.Address, m.Iface, err)
			continue
		}

		m.mu.Lock()
		changed := force || mac != m.mac
		if mac != m.mac {
			m.macSince = time.Now()
		}
		m.mac = mac
		m.mu.Unlock()

		if !changed {
			continue
		}
		if mac == "" {
			log.Printf("neigh: %s (%s) MAC утерян — снимаю запись adjacency", m.Name, m.Address)
			if err := a.applyNeigh(m); err != nil {
				log.Printf("neigh: %s: ошибка снятия записи: %v", m.Name, err)
			}
			continue
		}
		log.Printf("neigh: %s (%s@%s) -> %s", m.Name, m.Address, m.Iface, mac)
		if err := a.applyNeigh(m); err != nil {
			log.Printf("neigh: %s: ошибка применения: %v", m.Name, err)
		}
	}
}

// applyNeigh заливает точечный бандл для ОДНОГО члена: снимает прежнее
// правило adjacency (если было) и, если MAC резолвлен, ставит новое. Формат
// файла бандла — тот же, что и для таблицы слотов (FlowBundle в maglev.go).
func (a *Agent) applyNeigh(m *Member) error {
	m.mu.Lock()
	mac := m.mac
	m.mu.Unlock()

	table := a.cfg.AdjTable
	out := fmt.Sprintf("flow delete table=%d,ip,nw_dst=%s\n", table, m.Address)
	if mac != "" {
		out += fmt.Sprintf("flow add table=%d,priority=100,ip,nw_dst=%s actions=mod_dl_dst:%s,output:%d\n",
			table, m.Address, mac, m.OFPort)
	}

	path := fmt.Sprintf("/var/run/openvswitch/neigh-%s.bundle", m.Name)
	if err := os.WriteFile(path, []byte(out), 0o644); err != nil {
		return err
	}
	res, err := exec.Command("ovs-ofctl", "-O", a.cfg.OFVersion, "bundle", a.cfg.Bridge, path).CombinedOutput()
	if err != nil {
		return fmt.Errorf("ovs-ofctl bundle: %v: %s", err, res)
	}
	return nil
}