Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3a563a29ee | ||
|
|
734db5d779 | ||
|
|
36fc8ec8c6 | ||
|
|
eeeba6b7c0 |
+198
-5
@@ -1,10 +1,203 @@
|
||||
package main
|
||||
|
||||
import "net/http"
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httputil"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.cin.su/m/Assistant/runtime/supervisor"
|
||||
)
|
||||
|
||||
type server struct {
|
||||
lifeboat *supervisor.PersistentLifeboat
|
||||
stableUpstream string
|
||||
candidate string
|
||||
client *http.Client
|
||||
}
|
||||
|
||||
func main() {
|
||||
_ = http.ListenAndServe(":8088", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte("lifeboat-ready\n"))
|
||||
}))
|
||||
statePath := envOr("SUPERVISOR_STATE_PATH", "/var/lib/supervisor/release-state.json")
|
||||
stable := strings.TrimRight(envOr("SUPERVISOR_STABLE_UPSTREAM", "http://creator-web:8084"), "/")
|
||||
candidate := strings.TrimRight(envOr("SUPERVISOR_CANDIDATE_UPSTREAM", "http://creator-web-candidate:8084"), "/")
|
||||
store, err := supervisor.NewDurableStore(statePath)
|
||||
if err != nil {
|
||||
fatal(err)
|
||||
}
|
||||
lifeboat := supervisor.NewPersistentLifeboat(store)
|
||||
if _, err := lifeboat.Snapshot(); errors.Is(err, os.ErrNotExist) {
|
||||
baseline := supervisor.Release{ID: envOr("SUPERVISOR_BASELINE_RELEASE", "bootstrap"), Version: envOr("SUPERVISOR_BASELINE_RELEASE", "bootstrap")}
|
||||
if ok, startErr := lifeboat.Start(baseline); startErr != nil || !ok {
|
||||
fatal(fmt.Errorf("initialize baseline: %v", startErr))
|
||||
}
|
||||
if ok, observeErr := lifeboat.Record(supervisor.Observation{ReleaseID: baseline.ID, Ready: true, Reason: "supervisor baseline"}); observeErr != nil || !ok {
|
||||
fatal(fmt.Errorf("initialize baseline observation: %v", observeErr))
|
||||
}
|
||||
if ok, trafficErr := lifeboat.SetTraffic(supervisor.TrafficStable, stable, candidate); trafficErr != nil || !ok {
|
||||
fatal(fmt.Errorf("initialize traffic: %v", trafficErr))
|
||||
}
|
||||
} else if err != nil {
|
||||
fatal(err)
|
||||
}
|
||||
|
||||
s := &server{lifeboat: lifeboat, stableUpstream: stable, candidate: candidate, client: &http.Client{Timeout: 5 * time.Second}}
|
||||
public := http.NewServeMux()
|
||||
public.HandleFunc("GET /healthz", s.health)
|
||||
public.HandleFunc("GET /readyz", s.ready)
|
||||
public.HandleFunc("/", s.proxy)
|
||||
control := http.NewServeMux()
|
||||
control.HandleFunc("GET /v1/state", s.state)
|
||||
control.HandleFunc("POST /v1/releases/start", s.start)
|
||||
control.HandleFunc("POST /v1/releases/observe", s.observe)
|
||||
control.HandleFunc("POST /v1/releases/promote", s.promote)
|
||||
control.HandleFunc("POST /v1/releases/rollback", s.rollback)
|
||||
control.HandleFunc("POST /v1/traffic/canary", s.canary)
|
||||
control.HandleFunc("POST /v1/traffic/promote", s.promoteTraffic)
|
||||
control.HandleFunc("POST /v1/traffic/rollback", s.rollbackTraffic)
|
||||
|
||||
go func() {
|
||||
if err := http.ListenAndServe(envOr("SUPERVISOR_CONTROL_ADDR", ":8095"), control); err != nil {
|
||||
fatal(err)
|
||||
}
|
||||
}()
|
||||
if err := http.ListenAndServe(envOr("LISTEN_ADDR", ":8080"), public); err != nil {
|
||||
fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *server) health(w http.ResponseWriter, _ *http.Request) {
|
||||
writeJSON(w, map[string]string{"runtime": "supervisor", "status": "ok"})
|
||||
}
|
||||
|
||||
func (s *server) ready(w http.ResponseWriter, r *http.Request) {
|
||||
state, err := s.lifeboat.Snapshot()
|
||||
if err != nil || state.StableUpstream == "" {
|
||||
http.Error(w, "supervisor state unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
request, err := http.NewRequestWithContext(r.Context(), http.MethodGet, state.StableUpstream+"/healthz", nil)
|
||||
if err != nil {
|
||||
http.Error(w, "stable upstream unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
response, err := s.client.Do(request)
|
||||
if err != nil || response.StatusCode/100 != 2 {
|
||||
if response != nil {
|
||||
response.Body.Close()
|
||||
}
|
||||
http.Error(w, "stable upstream unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
response.Body.Close()
|
||||
writeJSON(w, map[string]string{"runtime": "supervisor", "status": "ready"})
|
||||
}
|
||||
|
||||
func (s *server) proxy(w http.ResponseWriter, r *http.Request) {
|
||||
state, err := s.lifeboat.Snapshot()
|
||||
if err != nil {
|
||||
http.Error(w, "supervisor state unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
target := state.StableUpstream
|
||||
selected := "stable"
|
||||
if state.TrafficMode == supervisor.TrafficCandidate || (state.TrafficMode == supervisor.TrafficCanaryHeader && strings.EqualFold(r.Header.Get("X-Assistant-Canary"), "true")) {
|
||||
target = state.CandidateUpstream
|
||||
selected = "candidate"
|
||||
}
|
||||
parsed, err := url.Parse(target)
|
||||
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
|
||||
http.Error(w, "invalid upstream", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
proxy := httputil.NewSingleHostReverseProxy(parsed)
|
||||
proxy.ModifyResponse = func(response *http.Response) error {
|
||||
response.Header.Set("X-Assistant-Traffic-Target", selected)
|
||||
return nil
|
||||
}
|
||||
proxy.ErrorHandler = func(writer http.ResponseWriter, _ *http.Request, _ error) {
|
||||
http.Error(writer, "upstream unavailable", http.StatusBadGateway)
|
||||
}
|
||||
proxy.ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
func (s *server) state(w http.ResponseWriter, _ *http.Request) {
|
||||
state, err := s.lifeboat.Snapshot()
|
||||
if err != nil {
|
||||
http.Error(w, "state unavailable", 503)
|
||||
return
|
||||
}
|
||||
writeJSON(w, state)
|
||||
}
|
||||
func (s *server) start(w http.ResponseWriter, r *http.Request) {
|
||||
var release supervisor.Release
|
||||
if !decode(w, r, &release) {
|
||||
return
|
||||
}
|
||||
ok, err := s.lifeboat.Start(release)
|
||||
respond(w, ok, err)
|
||||
}
|
||||
func (s *server) observe(w http.ResponseWriter, r *http.Request) {
|
||||
var observation supervisor.Observation
|
||||
if !decode(w, r, &observation) {
|
||||
return
|
||||
}
|
||||
ok, err := s.lifeboat.Record(observation)
|
||||
respond(w, ok, err)
|
||||
}
|
||||
func (s *server) promote(w http.ResponseWriter, _ *http.Request) {
|
||||
ok, err := s.lifeboat.Promote()
|
||||
respond(w, ok, err)
|
||||
}
|
||||
func (s *server) rollback(w http.ResponseWriter, _ *http.Request) {
|
||||
ok, err := s.lifeboat.Rollback()
|
||||
respond(w, ok, err)
|
||||
}
|
||||
func (s *server) canary(w http.ResponseWriter, _ *http.Request) {
|
||||
ok, err := s.lifeboat.SetTraffic(supervisor.TrafficCanaryHeader, s.stableUpstream, s.candidate)
|
||||
respond(w, ok, err)
|
||||
}
|
||||
func (s *server) promoteTraffic(w http.ResponseWriter, _ *http.Request) {
|
||||
ok, err := s.lifeboat.Promote()
|
||||
respond(w, ok, err)
|
||||
}
|
||||
func (s *server) rollbackTraffic(w http.ResponseWriter, _ *http.Request) {
|
||||
ok, err := s.lifeboat.RollbackTraffic()
|
||||
respond(w, ok, err)
|
||||
}
|
||||
|
||||
func decode(w http.ResponseWriter, r *http.Request, value any) bool {
|
||||
decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16<<10))
|
||||
decoder.DisallowUnknownFields()
|
||||
if err := decoder.Decode(value); err != nil {
|
||||
http.Error(w, "invalid control request", 400)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
func respond(w http.ResponseWriter, ok bool, err error) {
|
||||
if err != nil {
|
||||
http.Error(w, "control state unavailable", 503)
|
||||
return
|
||||
}
|
||||
if !ok {
|
||||
http.Error(w, "control transition rejected", 409)
|
||||
return
|
||||
}
|
||||
writeJSON(w, map[string]bool{"ok": true})
|
||||
}
|
||||
func writeJSON(w http.ResponseWriter, value any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(value)
|
||||
}
|
||||
func envOr(name, fallback string) string {
|
||||
if value := strings.TrimSpace(os.Getenv(name)); value != "" {
|
||||
return value
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
func fatal(err error) { fmt.Fprintln(os.Stderr, err); os.Exit(1) }
|
||||
|
||||
@@ -277,7 +277,7 @@ services:
|
||||
- --port
|
||||
- "8086"
|
||||
- --model
|
||||
- /models/Qwen3-8B-Q4_K_M.gguf
|
||||
- /models/Qwen3.8-27B-UD-Q2_K_XL.gguf
|
||||
- --alias
|
||||
- domovoy-lifeboat
|
||||
- --jinja
|
||||
@@ -299,6 +299,29 @@ services:
|
||||
timeout: 3s
|
||||
retries: 10
|
||||
|
||||
supervisor:
|
||||
build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: supervisor, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } }
|
||||
restart: unless-stopped
|
||||
environment:
|
||||
LISTEN_ADDR: ":8080"
|
||||
SUPERVISOR_CONTROL_ADDR: ":8095"
|
||||
SUPERVISOR_STATE_PATH: /var/lib/supervisor/release-state.json
|
||||
SUPERVISOR_STABLE_UPSTREAM: http://creator-web:8084
|
||||
SUPERVISOR_CANDIDATE_UPSTREAM: http://creator-web-candidate:8084
|
||||
SUPERVISOR_BASELINE_RELEASE: ${ASSISTANT_SOURCE_REVISION:-unknown}
|
||||
ports:
|
||||
- "${CREATOR_WEB_PORT:-18080}:8080"
|
||||
- "127.0.0.1:${SUPERVISOR_CONTROL_PORT:-18085}:8095"
|
||||
depends_on:
|
||||
creator-web: { condition: service_healthy }
|
||||
volumes:
|
||||
- assistant_supervisor_state:/var/lib/supervisor
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null && wget -qO- http://localhost:8080/readyz >/dev/null"]
|
||||
interval: 10s
|
||||
timeout: 5s
|
||||
retries: 20
|
||||
|
||||
creator-web:
|
||||
build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: creator-web, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } }
|
||||
restart: unless-stopped
|
||||
@@ -310,7 +333,6 @@ services:
|
||||
COGNITIVE_RUNTIME_URL: http://cognitive-worker:8083
|
||||
AGENT_WORKER_URL: http://agent-worker:8090
|
||||
ASSISTANT_BUILD_REVISION: ${ASSISTANT_SOURCE_REVISION:-unknown}
|
||||
ports: ["${CREATOR_WEB_PORT:-18080}:8084"]
|
||||
depends_on: [core-runtime, continuity-runtime, commit-runtime, cognitive-worker]
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "wget -qO- http://localhost:8084/healthz >/dev/null"]
|
||||
@@ -318,6 +340,12 @@ services:
|
||||
timeout: 3s
|
||||
retries: 10
|
||||
|
||||
creator-web-candidate:
|
||||
profiles: ["assistant-canary"]
|
||||
extends:
|
||||
service: creator-web
|
||||
restart: "no"
|
||||
|
||||
volumes:
|
||||
assistant_pg:
|
||||
assistant_nats:
|
||||
@@ -333,6 +361,7 @@ volumes:
|
||||
external: true
|
||||
assistant_research_sources:
|
||||
name: assistant_research_sources
|
||||
assistant_supervisor_state:
|
||||
|
||||
networks:
|
||||
llm-local:
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
BASE_DIR="${REPO_DIR:-$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)}"
|
||||
ENV_FILE="${ENV_FILE:-/home/m/assistant-deploy/production-compose.env}"
|
||||
COMPOSE_FILE="$BASE_DIR/deploy/docker-compose.prod.yml"
|
||||
SUPERVISOR_CONTROL_URL="${SUPERVISOR_CONTROL_URL:-http://127.0.0.1:${SUPERVISOR_CONTROL_PORT:-18085}}"
|
||||
REVISION="${1:?candidate revision is required}"
|
||||
|
||||
set -a
|
||||
. "$ENV_FILE"
|
||||
set +a
|
||||
|
||||
compose() { docker compose --env-file "$ENV_FILE" -f "$COMPOSE_FILE" "$@"; }
|
||||
control() { curl -fsS --max-time 10 -X POST "$SUPERVISOR_CONTROL_URL$1" "${@:2}" >/dev/null; }
|
||||
fail_rollback() {
|
||||
control /v1/traffic/rollback || true
|
||||
control /v1/releases/rollback || true
|
||||
echo "production canary failed; stable traffic rollback requested" >&2
|
||||
exit 1
|
||||
}
|
||||
|
||||
compose --profile assistant-canary build --build-arg "SOURCE_REVISION=$REVISION" creator-web-candidate
|
||||
compose --profile assistant-canary up -d --no-build creator-web-candidate
|
||||
for attempt in $(seq 1 30); do
|
||||
status="$(docker inspect -f '{{if .State.Health}}{{.State.Health.Status}}{{else}}unknown{{end}}' assistant-creator-web-candidate-1 2>/dev/null || true)"
|
||||
[[ "$status" == healthy ]] && break
|
||||
[[ "$attempt" == 30 ]] && fail_rollback
|
||||
sleep 2
|
||||
done
|
||||
|
||||
control /v1/releases/start -H 'Content-Type: application/json' --data "{\"id\":\"candidate-$REVISION\",\"version\":\"$REVISION\"}" || fail_rollback
|
||||
control /v1/releases/observe -H 'Content-Type: application/json' --data "{\"release_id\":\"candidate-$REVISION\",\"ready\":true,\"reason\":\"candidate healthcheck passed\",\"observed_at_unix_millis\":$(date +%s%3N)}" || fail_rollback
|
||||
control /v1/traffic/canary || fail_rollback
|
||||
|
||||
headers="$(mktemp)"
|
||||
trap 'rm -f "$headers"' EXIT
|
||||
if ! curl -fsS --max-time 10 -D "$headers" -H 'X-Assistant-Canary: true' "https://assistant.cin.su/login" -o /dev/null; then fail_rollback; fi
|
||||
grep -qi '^X-Assistant-Traffic-Target: candidate' "$headers" || fail_rollback
|
||||
|
||||
control /v1/releases/promote || fail_rollback
|
||||
echo "production canary promoted revision=$REVISION"
|
||||
@@ -1 +1 @@
|
||||
{"task_file":"dev/tasks/GEN2-027-production-canary-and-controlled-gen2-traffic-transition.json","task_id":"GEN2-027","task_revision":1}
|
||||
{"task_file":"dev/tasks/GEN2-DEPLOY-001-supervisor-and-production-rollback.json","task_id":"GEN2-DEPLOY-001","task_revision":1}
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"task_id": "GEN2-DEPLOY-001",
|
||||
"task_revision": 1,
|
||||
"architecture_revision": "ARCH-GEN2-000",
|
||||
"base_repo_revision": "gen2-integration",
|
||||
"risk_tier": "E",
|
||||
"intent": "Развернуть независимый Supervisor control plane с durable release state, controlled traffic switching и доказуемым rollback перед GEN2-027.",
|
||||
"requirement_ids": ["REQ-GEN2-DEPLOY-001-SUPERVISOR", "REQ-GEN2-DEPLOY-001-TRAFFIC", "REQ-GEN2-DEPLOY-001-ROLLBACK", "REQ-GEN2-DEPLOY-001-LIFEBOAT"],
|
||||
"allowed_change_scope": {
|
||||
"modules": ["deployment control", "runtime supervisor", "runtime health"],
|
||||
"paths": [
|
||||
"deploy/**",
|
||||
"runtime/supervisor/**",
|
||||
"runtime/health/**",
|
||||
"cmd/supervisor/**",
|
||||
"dev/ACTIVE_TASK.json",
|
||||
"dev/tasks/GEN2-DEPLOY-001-supervisor-and-production-rollback.json",
|
||||
"docs/evidence/**"
|
||||
]
|
||||
},
|
||||
"forbidden_change_scope": {
|
||||
"paths": ["GEN2-028", "database/**", "security/**", "architecture/**", "Authority/Commit bypass", "cognitive-lifeboat as Supervisor"]
|
||||
},
|
||||
"allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
|
||||
"contracts": {"consume": ["GEN2-027 production canary contract", "runtime/supervisor durable state", "existing deployment update controller"], "may_change": ["deployment topology, Supervisor HTTP control contract and traffic routing only"]},
|
||||
"semantic_owner": "Operations/Supervisor; not a semantic owner and not a cognitive provider.",
|
||||
"canonical_state": ["durable release/current/last-known-good evidence owned by Supervisor control plane", "traffic target is Supervisor-owned operational state"],
|
||||
"state_transitions": ["baseline -> candidate -> observed healthy -> header canary -> promoted OR independent rollback to last-known-good"],
|
||||
"side_effects_and_privileges": ["Control API is bound only to the production-internal loopback port.", "Supervisor has no database, Authority or Commit credentials.", "Candidate traffic is explicit and rollback remains available."],
|
||||
"persistence_and_migrations": ["No database changes; durable state is stored in a dedicated named volume outside candidate containers."],
|
||||
"failure_retry_restart": ["Supervisor restart reloads release and traffic state from its volume.", "Candidate failure must not erase last-known-good or block stable traffic rollback.", "Lifeboat model failure remains a separately reported degraded provider condition."],
|
||||
"assumptions": ["GEN2-027 resumes only after this task's production evidence; GEN2-028 is not started.", "The existing public ingress forwards to the configured creator web host port."],
|
||||
"acceptance_scenarios": ["GEN2-DEPLOY-001-SCN-01: Supervisor is a separate healthy production container", "GEN2-DEPLOY-001-SCN-02: release state survives Supervisor restart", "GEN2-DEPLOY-001-SCN-03: canary header reaches candidate while default traffic stays stable", "GEN2-DEPLOY-001-SCN-04: candidate failure rolls traffic back to last-known-good", "GEN2-DEPLOY-001-SCN-05: cognitive lifeboat diagnosis and degraded role are explicit", "GEN2-DEPLOY-001-EXIT: GEN2-027 rollback prerequisites are proven"],
|
||||
"required_tests": ["FAST: go test ./runtime/supervisor/... ./runtime/health/... ./cmd/supervisor/...", "FAST: go vet ./runtime/supervisor/... ./runtime/health/... ./cmd/supervisor/...", "FAST governance: preflight", "FAST scope: check-diff", "FAST independent: independent", "FAST whitespace: git diff --check", "TARGETED recovery: Supervisor restart, canary and rollback on docker.cin.su"],
|
||||
"terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_CONFLICT", "REQUIRES_MIGRATION_DECISION"]
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// DurableStore is supervisor-owned external control-plane state. It is not a
|
||||
@@ -63,12 +64,17 @@ func (s *DurableStore) Load() (Lifeboat, error) {
|
||||
}
|
||||
|
||||
// PersistentLifeboat reloads before every transition and commits after it.
|
||||
type PersistentLifeboat struct{ store *DurableStore }
|
||||
type PersistentLifeboat struct {
|
||||
store *DurableStore
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func NewPersistentLifeboat(store *DurableStore) *PersistentLifeboat {
|
||||
return &PersistentLifeboat{store: store}
|
||||
}
|
||||
func (p *PersistentLifeboat) transition(fn func(*Lifeboat) bool) (bool, error) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
value, err := p.store.Load()
|
||||
if err != nil && !os.IsNotExist(err) {
|
||||
return false, err
|
||||
@@ -91,4 +97,30 @@ func (p *PersistentLifeboat) Record(observation Observation) (bool, error) {
|
||||
func (p *PersistentLifeboat) Rollback() (bool, error) {
|
||||
return p.transition(func(value *Lifeboat) bool { return value.Rollback() })
|
||||
}
|
||||
func (p *PersistentLifeboat) Snapshot() (Lifeboat, error) { return p.store.Load() }
|
||||
func (p *PersistentLifeboat) SetTraffic(mode TrafficMode, stableUpstream, candidateUpstream string) (bool, error) {
|
||||
return p.transition(func(value *Lifeboat) bool { return value.SetTraffic(mode, stableUpstream, candidateUpstream) })
|
||||
}
|
||||
|
||||
func (p *PersistentLifeboat) Promote() (bool, error) {
|
||||
return p.transition(func(value *Lifeboat) bool {
|
||||
if value.Current.State != ReleaseHealthy || value.Current.ID == "" {
|
||||
return false
|
||||
}
|
||||
return value.SetTraffic(TrafficCandidate, value.StableUpstream, value.CandidateUpstream)
|
||||
})
|
||||
}
|
||||
|
||||
func (p *PersistentLifeboat) RollbackTraffic() (bool, error) {
|
||||
return p.transition(func(value *Lifeboat) bool {
|
||||
if !value.Rollback() {
|
||||
return false
|
||||
}
|
||||
return value.SetTraffic(TrafficStable, value.StableUpstream, value.CandidateUpstream)
|
||||
})
|
||||
}
|
||||
|
||||
func (p *PersistentLifeboat) Snapshot() (Lifeboat, error) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return p.store.Load()
|
||||
}
|
||||
|
||||
@@ -9,6 +9,14 @@ const (
|
||||
ReleaseRolledBack ReleaseState = "ROLLED_BACK"
|
||||
)
|
||||
|
||||
type TrafficMode string
|
||||
|
||||
const (
|
||||
TrafficStable TrafficMode = "STABLE"
|
||||
TrafficCanaryHeader TrafficMode = "CANARY_HEADER"
|
||||
TrafficCandidate TrafficMode = "CANDIDATE"
|
||||
)
|
||||
|
||||
type Release struct {
|
||||
ID string
|
||||
Version string
|
||||
@@ -22,9 +30,12 @@ type Observation struct {
|
||||
ObservedAtUnixMillis int64
|
||||
}
|
||||
type Lifeboat struct {
|
||||
Current Release
|
||||
LastKnownGood Release
|
||||
Observations []Observation
|
||||
Current Release
|
||||
LastKnownGood Release
|
||||
Observations []Observation
|
||||
TrafficMode TrafficMode
|
||||
StableUpstream string
|
||||
CandidateUpstream string
|
||||
}
|
||||
|
||||
func (l *Lifeboat) Start(candidate Release) bool {
|
||||
@@ -60,3 +71,16 @@ func (l *Lifeboat) Rollback() bool {
|
||||
l.Current.State = ReleaseRolledBack
|
||||
return true
|
||||
}
|
||||
|
||||
func (l *Lifeboat) SetTraffic(mode TrafficMode, stableUpstream, candidateUpstream string) bool {
|
||||
if mode != TrafficStable && mode != TrafficCanaryHeader && mode != TrafficCandidate {
|
||||
return false
|
||||
}
|
||||
if stableUpstream == "" || (mode != TrafficStable && candidateUpstream == "") {
|
||||
return false
|
||||
}
|
||||
l.TrafficMode = mode
|
||||
l.StableUpstream = stableUpstream
|
||||
l.CandidateUpstream = candidateUpstream
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -47,3 +47,38 @@ func TestPersistentLifeboatRestoresLastKnownGoodAfterCandidateFailure(t *testing
|
||||
t.Fatalf("state=%+v", state)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPersistentSupervisorRestoresTrafficModeAfterRestart(t *testing.T) {
|
||||
store, err := supervisor.NewDurableStore(filepath.Join(t.TempDir(), "control", "release-state.json"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
first := supervisor.NewPersistentLifeboat(store)
|
||||
if ok, err := first.Start(supervisor.Release{ID: "candidate", Version: "2"}); err != nil || !ok {
|
||||
t.Fatalf("start: %v %v", ok, err)
|
||||
}
|
||||
if ok, err := first.Record(supervisor.Observation{ReleaseID: "candidate", Ready: true, ObservedAtUnixMillis: 1}); err != nil || !ok {
|
||||
t.Fatalf("observe: %v %v", ok, err)
|
||||
}
|
||||
if ok, err := first.SetTraffic(supervisor.TrafficCanaryHeader, "http://stable:8084", "http://candidate:8084"); err != nil || !ok {
|
||||
t.Fatalf("canary: %v %v", ok, err)
|
||||
}
|
||||
second := supervisor.NewPersistentLifeboat(store)
|
||||
state, err := second.Snapshot()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if state.TrafficMode != supervisor.TrafficCanaryHeader || state.CandidateUpstream != "http://candidate:8084" || state.LastKnownGood.ID != "candidate" {
|
||||
t.Fatalf("state=%+v", state)
|
||||
}
|
||||
if ok, err := second.RollbackTraffic(); err != nil || !ok {
|
||||
t.Fatalf("rollback traffic: %v %v", ok, err)
|
||||
}
|
||||
state, err = second.Snapshot()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if state.TrafficMode != supervisor.TrafficStable || state.Current.State != supervisor.ReleaseRolledBack {
|
||||
t.Fatalf("rollback state=%+v", state)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user