feat: complete daemon operations and recovery tooling
This commit is contained in:
@@ -0,0 +1,291 @@
|
||||
package backendstatus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"io"
|
||||
"log/slog"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"yms-daemon/internal/containerengine"
|
||||
"yms-daemon/internal/daemonapi"
|
||||
"yms-daemon/internal/deploymentconfig"
|
||||
"yms-daemon/internal/hostnginx"
|
||||
"yms-daemon/internal/transaction"
|
||||
)
|
||||
|
||||
// TestContainerDiagnoseUndeployed 验证全新机器（无部署记录、无容器、无历史事务）被诊断为健康且未部署。
|
||||
func TestContainerDiagnoseUndeployed(t *testing.T) {
|
||||
d, _ := newContainerDiagnoser(t, &fakeEngine{}, &fakeGateway{port: 8080})
|
||||
diagnosis, err := d.Diagnose(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("diagnose undeployed container backend: %v", err)
|
||||
}
|
||||
if !diagnosis.Healthy {
|
||||
t.Fatalf("undeployed machine should be healthy: %+v", diagnosis)
|
||||
}
|
||||
if diagnosis.Items[0].Level != daemonapi.DiagnosisLevelOK || diagnosis.Items[0].Code != codeUndeployed {
|
||||
t.Fatalf("unexpected undeployed finding: %+v", diagnosis.Items)
|
||||
}
|
||||
}
|
||||
|
||||
// TestContainerDiagnoseHealthy 验证部署记录、活动容器与 Nginx 三者一致时被诊断为健康。
|
||||
func TestContainerDiagnoseHealthy(t *testing.T) {
|
||||
engine := &fakeEngine{containers: map[string]containerengine.Container{
|
||||
"backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true},
|
||||
}}
|
||||
d, store := newContainerDiagnoser(t, engine, &fakeGateway{port: 8080})
|
||||
commitDeployment(t, store, transaction.BackendContainerDeployment{
|
||||
ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080",
|
||||
})
|
||||
|
||||
diagnosis, err := d.Diagnose(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("diagnose healthy container backend: %v", err)
|
||||
}
|
||||
if !diagnosis.Healthy {
|
||||
t.Fatalf("healthy container backend reported drift: %+v", diagnosis)
|
||||
}
|
||||
for _, item := range diagnosis.Items {
|
||||
if item.Level == daemonapi.DiagnosisLevelDrift || item.Level == daemonapi.DiagnosisLevelFixable {
|
||||
t.Fatalf("unexpected non-ok finding: %+v", item)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestContainerDiagnoseNginxWrongPort 验证 Nginx 指向错误端口而活动容器健康时判定为可自动修复。
|
||||
func TestContainerDiagnoseNginxWrongPort(t *testing.T) {
|
||||
engine := &fakeEngine{containers: map[string]containerengine.Container{
|
||||
"backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true},
|
||||
}}
|
||||
d, store := newContainerDiagnoser(t, engine, &fakeGateway{port: 8081})
|
||||
commitDeployment(t, store, transaction.BackendContainerDeployment{
|
||||
ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080",
|
||||
})
|
||||
|
||||
diagnosis, err := d.Diagnose(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("diagnose nginx wrong port: %v", err)
|
||||
}
|
||||
if !diagnosis.Healthy {
|
||||
t.Fatalf("nginx wrong port with healthy container should be fixable: %+v", diagnosis)
|
||||
}
|
||||
if !hasFinding(diagnosis, daemonapi.DiagnosisLevelFixable, codeNginxWrongPort) {
|
||||
t.Fatalf("missing nginx wrong port finding: %+v", diagnosis.Items)
|
||||
}
|
||||
}
|
||||
|
||||
// TestContainerDiagnoseActiveContainerMissing 验证部署记录的活动容器缺失时判定为漂移。
|
||||
func TestContainerDiagnoseActiveContainerMissing(t *testing.T) {
|
||||
d, store := newContainerDiagnoser(t, &fakeEngine{}, &fakeGateway{port: 8080})
|
||||
commitDeployment(t, store, transaction.BackendContainerDeployment{
|
||||
ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080",
|
||||
})
|
||||
|
||||
diagnosis, err := d.Diagnose(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("diagnose missing active container: %v", err)
|
||||
}
|
||||
if diagnosis.Healthy {
|
||||
t.Fatalf("missing active container should be drift: %+v", diagnosis)
|
||||
}
|
||||
if !hasFinding(diagnosis, daemonapi.DiagnosisLevelDrift, codeActiveContainerMissing) {
|
||||
t.Fatalf("missing active container finding: %+v", diagnosis.Items)
|
||||
}
|
||||
}
|
||||
|
||||
// TestContainerDiagnoseInactiveResidual 验证非活动槽位残留已停止容器时判定为可自动修复。
|
||||
func TestContainerDiagnoseInactiveResidual(t *testing.T) {
|
||||
engine := &fakeEngine{containers: map[string]containerengine.Container{
|
||||
"backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true},
|
||||
"backend-8081": {ID: "container-8081", Name: "backend-8081", Running: false},
|
||||
}}
|
||||
d, store := newContainerDiagnoser(t, engine, &fakeGateway{port: 8080})
|
||||
commitDeployment(t, store, transaction.BackendContainerDeployment{
|
||||
ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080",
|
||||
})
|
||||
|
||||
diagnosis, err := d.Diagnose(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("diagnose inactive residual: %v", err)
|
||||
}
|
||||
if !hasFinding(diagnosis, daemonapi.DiagnosisLevelFixable, codeInactiveContainerResidual) {
|
||||
t.Fatalf("missing inactive residual finding: %+v", diagnosis.Items)
|
||||
}
|
||||
}
|
||||
|
||||
// TestContainerDiagnoseContainerWithoutRecord 验证无部署记录但现场存在容器时判定为漂移。
|
||||
func TestContainerDiagnoseContainerWithoutRecord(t *testing.T) {
|
||||
engine := &fakeEngine{containers: map[string]containerengine.Container{
|
||||
"backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true},
|
||||
}}
|
||||
d, _ := newContainerDiagnoser(t, engine, &fakeGateway{port: 8080})
|
||||
|
||||
diagnosis, err := d.Diagnose(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("diagnose container without record: %v", err)
|
||||
}
|
||||
if diagnosis.Healthy {
|
||||
t.Fatalf("container without record should be drift: %+v", diagnosis)
|
||||
}
|
||||
if !hasFinding(diagnosis, daemonapi.DiagnosisLevelDrift, codeContainerWithoutRecord) {
|
||||
t.Fatalf("missing container-without-record finding: %+v", diagnosis.Items)
|
||||
}
|
||||
}
|
||||
|
||||
// TestContainerReconcileApplyFixesNginx 验证 reconcile --apply 能把指向错误的 Nginx 切回记录端口。
|
||||
func TestContainerReconcileApplyFixesNginx(t *testing.T) {
|
||||
engine := &fakeEngine{containers: map[string]containerengine.Container{
|
||||
"backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true},
|
||||
}}
|
||||
gateway := &fakeGateway{port: 8081}
|
||||
d, store := newContainerDiagnoser(t, engine, gateway)
|
||||
commitDeployment(t, store, transaction.BackendContainerDeployment{
|
||||
ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080",
|
||||
})
|
||||
|
||||
diagnosis, err := d.Reconcile(context.Background(), true)
|
||||
if err != nil {
|
||||
t.Fatalf("reconcile apply nginx: %v", err)
|
||||
}
|
||||
if gateway.port != 8080 {
|
||||
t.Fatalf("reconcile did not switch Nginx to port 8080: port=%d", gateway.port)
|
||||
}
|
||||
if !diagnosis.Healthy {
|
||||
t.Fatalf("reconcile result should be healthy: %+v", diagnosis)
|
||||
}
|
||||
if hasFinding(diagnosis, daemonapi.DiagnosisLevelFixable, codeNginxWrongPort) {
|
||||
t.Fatalf("nginx wrong port should be fixed after reconcile: %+v", diagnosis.Items)
|
||||
}
|
||||
}
|
||||
|
||||
// hasFinding 判断诊断结果中是否存在指定级别与稳定标识的诊断项。
|
||||
func hasFinding(diagnosis daemonapi.Diagnosis, level string, code string) bool {
|
||||
for _, item := range diagnosis.Items {
|
||||
if item.Level == level && item.Code == code {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// newContainerDiagnoser 构造 container 后端的诊断器及其底层存储，返回诊断器与可写的存储。
|
||||
func newContainerDiagnoser(t *testing.T, engine *fakeEngine, gateway *fakeGateway) (*Diagnoser, *transaction.Store) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
store, err := transaction.OpenStore(ctx, filepath.Join(t.TempDir(), "transactions.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("open transaction store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = store.Close() })
|
||||
coordinator, err := transaction.NewCoordinator(store, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
if err != nil {
|
||||
t.Fatalf("create transaction coordinator: %v", err)
|
||||
}
|
||||
config := deploymentconfig.Config{
|
||||
Daemon: deploymentconfig.Daemon{Environment: deploymentconfig.EnvironmentDev},
|
||||
Backend: deploymentconfig.Backend{
|
||||
Type: deploymentconfig.BackendTypeContainer,
|
||||
SystemctlPath: "/bin/systemctl",
|
||||
Slot: deploymentconfig.BackendSlots{
|
||||
Port8080: deploymentconfig.BackendSlot{ContainerName: "backend-8080", HealthEndpoint: "http://127.0.0.1:8080/yms/actuator/health"},
|
||||
Port8081: deploymentconfig.BackendSlot{ContainerName: "backend-8081", HealthEndpoint: "http://127.0.0.1:8081/yms/actuator/health"},
|
||||
},
|
||||
},
|
||||
}
|
||||
diagnoser, err := New(config, store, coordinator, engine, nil, gateway)
|
||||
if err != nil {
|
||||
t.Fatalf("create diagnoser: %v", err)
|
||||
}
|
||||
return diagnoser, store
|
||||
}
|
||||
|
||||
// commitDeployment 通过合法的状态机推进写入一条 container 部署记录。
|
||||
func commitDeployment(t *testing.T, store *transaction.Store, deployment transaction.BackendContainerDeployment) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
transactionID := "deploy-" + rand.Text()
|
||||
if _, _, err := store.CreateTransaction(ctx, transaction.CreateRequest{
|
||||
ID: transactionID,
|
||||
IdempotencyKey: "backend:container:test:" + rand.Text(),
|
||||
Source: "test",
|
||||
Service: "backend",
|
||||
}); err != nil {
|
||||
t.Fatalf("create deployment transaction: %v", err)
|
||||
}
|
||||
for _, state := range []transaction.State{
|
||||
transaction.StateValidating,
|
||||
transaction.StatePrepared,
|
||||
transaction.StateStarting,
|
||||
transaction.StateSwitching,
|
||||
transaction.StateVerifying,
|
||||
transaction.StateDraining,
|
||||
} {
|
||||
if _, err := store.Transition(ctx, transactionID, state, "test"); err != nil {
|
||||
t.Fatalf("advance deployment transaction: %v", err)
|
||||
}
|
||||
}
|
||||
if _, err := store.CommitBackendContainerDeployment(ctx, transactionID, deployment, "test committed"); err != nil {
|
||||
t.Fatalf("commit deployment: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// fakeGateway gateway 接口的内存实现，记录活动端口并在 Switch 时直接更新。
|
||||
type fakeGateway struct {
|
||||
port int
|
||||
content []byte
|
||||
}
|
||||
|
||||
// Read 返回当前快照与活动端口。
|
||||
func (g *fakeGateway) Read() (hostnginx.Snapshot, error) {
|
||||
content := g.content
|
||||
if content == nil {
|
||||
content = []byte("http {\n upstream yms-server {\n # yms-update managed upstream begin\n server 127.0.0.1:8080 max_fails=1 fail_timeout=2s;\n server 127.0.0.1:8081 max_fails=1 fail_timeout=2s;\n # yms-update managed upstream end\n }\n}\n")
|
||||
}
|
||||
return hostnginx.Snapshot{Content: append([]byte(nil), content...), ActivePort: g.port}, nil
|
||||
}
|
||||
|
||||
// Switch 直接更新活动端口并返回切换前快照。
|
||||
func (g *fakeGateway) Switch(_ context.Context, port int) (hostnginx.Snapshot, error) {
|
||||
previous, _ := g.Read()
|
||||
g.port = port
|
||||
return previous, nil
|
||||
}
|
||||
|
||||
// fakeEngine containerengine.Engine 的内存实现，维护容器表。
|
||||
type fakeEngine struct {
|
||||
containers map[string]containerengine.Container
|
||||
removed []string
|
||||
}
|
||||
|
||||
func (e *fakeEngine) Ping(context.Context) error { return nil }
|
||||
func (e *fakeEngine) PullImage(context.Context, string) error { return nil }
|
||||
func (e *fakeEngine) LoadImage(context.Context, io.Reader) error { return nil }
|
||||
func (e *fakeEngine) InspectImage(context.Context, string) (containerengine.Image, error) {
|
||||
return containerengine.Image{}, nil
|
||||
}
|
||||
func (e *fakeEngine) CreateContainer(context.Context, containerengine.ContainerSpec) (containerengine.Container, error) {
|
||||
return containerengine.Container{}, nil
|
||||
}
|
||||
func (e *fakeEngine) StartContainer(context.Context, string) error { return nil }
|
||||
func (e *fakeEngine) ContainerLogs(context.Context, string) (io.ReadCloser, error) {
|
||||
return io.NopCloser(strings.NewReader("")), nil
|
||||
}
|
||||
func (e *fakeEngine) StopContainer(context.Context, string, int) error { return nil }
|
||||
func (e *fakeEngine) InspectContainer(_ context.Context, name string) (containerengine.Container, error) {
|
||||
record, found := e.containers[name]
|
||||
if !found {
|
||||
return containerengine.Container{}, containerengine.ErrNotFound
|
||||
}
|
||||
return record, nil
|
||||
}
|
||||
func (e *fakeEngine) RemoveContainer(_ context.Context, name string, _ bool) error {
|
||||
delete(e.containers, name)
|
||||
e.removed = append(e.removed, name)
|
||||
return nil
|
||||
}
|
||||
func (e *fakeEngine) Close() error { return nil }
|
||||
|
||||
var _ containerengine.Engine = (*fakeEngine)(nil)
|
||||
Reference in New Issue
Block a user