451 lines
12 KiB
Go
451 lines
12 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"flag"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"crit.rip/gapmind/services/orchestra/service"
|
|
)
|
|
|
|
type Target struct {
|
|
Host string `json:"host"`
|
|
User string `json:"user"`
|
|
Service string `json:"service"`
|
|
Binary string `json:"binary"`
|
|
RemoteDir string `json:"remoteDir"`
|
|
CentralURL string `json:"centralURL"`
|
|
Token string `json:"token"`
|
|
SSHPassword string `json:"sshPassword"`
|
|
PublicURL string `json:"publicURL"`
|
|
Latitude string `json:"latitude"`
|
|
Longitude string `json:"longitude"`
|
|
TimeoutSec int `json:"timeoutSeconds"`
|
|
DHTListenPort int `json:"dhtListenPort"`
|
|
DHTPassive *bool `json:"dhtPassive"`
|
|
RSSInterval string `json:"rssInterval"`
|
|
Trackers string `json:"trackers"`
|
|
RSSFeeds string `json:"rssFeeds"`
|
|
DHTAPIToken string `json:"dhtApiToken"`
|
|
}
|
|
|
|
func (t Target) nodeId() string {
|
|
if t.Service == "dht" {
|
|
port := t.DHTListenPort
|
|
if port == 0 {
|
|
port = 6881
|
|
}
|
|
return fmt.Sprintf("%s:%d", t.Host, port)
|
|
}
|
|
return t.Host
|
|
}
|
|
|
|
func (t Target) resolvedNodeId() string {
|
|
host := sshHostname(t)
|
|
if host == "" {
|
|
host = t.Host
|
|
}
|
|
if t.Service == "dht" {
|
|
port := t.DHTListenPort
|
|
if port == 0 {
|
|
port = 6881
|
|
}
|
|
return fmt.Sprintf("%s:%d", host, port)
|
|
}
|
|
return host
|
|
}
|
|
|
|
func sshClientFor(t Target) (*service.SSHClient, error) {
|
|
remote := t.Host
|
|
if t.User != "" {
|
|
remote = t.User + "@" + t.Host
|
|
}
|
|
password := t.SSHPassword
|
|
if password == "" {
|
|
password = t.Token
|
|
}
|
|
client, err := service.NewSSHClient(remote, password)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if t.TimeoutSec > 0 {
|
|
client.SetTimeout(time.Duration(t.TimeoutSec) * time.Second)
|
|
}
|
|
return client, nil
|
|
}
|
|
|
|
func sshHostname(t Target) string {
|
|
client, err := sshClientFor(t)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
defer client.Close()
|
|
out, err := client.Run("ssh", client.Remote(), "hostname", "-s")
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
host := strings.TrimSpace(string(out))
|
|
if host == "" {
|
|
return ""
|
|
}
|
|
return strings.ToLower(host)
|
|
}
|
|
|
|
type Config struct {
|
|
Targets []Target `json:"targets"`
|
|
}
|
|
|
|
type peerUpdate struct {
|
|
NodeId string `json:"nodeId"`
|
|
Metadata map[string]interface{} `json:"metadata"`
|
|
Address string `json:"address,omitempty"`
|
|
Version string `json:"version,omitempty"`
|
|
Status string `json:"status,omitempty"`
|
|
Kind string `json:"kind,omitempty"`
|
|
}
|
|
|
|
func loadConfig(path string) Config {
|
|
data, err := os.ReadFile(path)
|
|
if err != nil {
|
|
fail(err)
|
|
}
|
|
var config Config
|
|
if err := json.Unmarshal(data, &config); err != nil {
|
|
fail(err)
|
|
}
|
|
return config
|
|
}
|
|
|
|
func filterTargets(targets []Target, only string) []Target {
|
|
if only == "" {
|
|
return targets
|
|
}
|
|
filter := map[string]bool{}
|
|
for _, s := range strings.Split(only, ",") {
|
|
filter[strings.TrimSpace(s)] = true
|
|
}
|
|
var out []Target
|
|
for _, t := range targets {
|
|
if filter[t.Service] {
|
|
out = append(out, t)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func main() {
|
|
configPath := flag.String("config", "orchestra.json", "path to deployment config")
|
|
only := flag.String("service", "", "filter to services, comma-separated (e.g. dht,pathfinder)")
|
|
flag.Parse()
|
|
|
|
args := flag.Args()
|
|
if len(args) == 0 {
|
|
fail(fmt.Errorf("usage: orchestra [flags] deploy|sync|nuke"))
|
|
}
|
|
subcommand := args[0]
|
|
config := loadConfig(*configPath)
|
|
targets := filterTargets(config.Targets, *only)
|
|
|
|
switch subcommand {
|
|
case "deploy":
|
|
cmdDeploy(targets)
|
|
case "sync":
|
|
fs := flag.NewFlagSet("sync", flag.ExitOnError)
|
|
restart := fs.Bool("restart", false, "restart remote services after metadata sync")
|
|
fs.Parse(args[1:])
|
|
cmdSync(targets, *restart)
|
|
case "nuke":
|
|
fs := flag.NewFlagSet("nuke", flag.ExitOnError)
|
|
confirm := fs.Bool("yes", false, "confirm destructive nuke without prompting")
|
|
redeploy := fs.Bool("redeploy", true, "redeploy all targets after nuking")
|
|
fs.Parse(args[1:])
|
|
cmdNuke(targets, *confirm, *redeploy)
|
|
default:
|
|
fail(fmt.Errorf("unknown subcommand %q: use deploy, sync, or nuke", subcommand))
|
|
}
|
|
}
|
|
|
|
func cmdDeploy(targets []Target) {
|
|
for _, target := range targets {
|
|
deploy(target)
|
|
}
|
|
}
|
|
|
|
func cmdNuke(targets []Target, confirm, redeploy bool) {
|
|
if len(targets) == 0 {
|
|
fmt.Fprintln(os.Stderr, "no targets to nuke")
|
|
return
|
|
}
|
|
if !confirm {
|
|
fmt.Fprintf(os.Stderr, "nuke: will stop + wipe %d targets and local caches. Re-run with --yes to confirm.\n", len(targets))
|
|
os.Exit(2)
|
|
}
|
|
for _, t := range targets {
|
|
nukeRemote(t)
|
|
}
|
|
nukeLocal()
|
|
if redeploy {
|
|
fmt.Println("nuke: redeploying network")
|
|
cmdDeploy(targets)
|
|
}
|
|
fmt.Println("nuke: done")
|
|
}
|
|
|
|
func nukeRemote(t Target) {
|
|
client, err := sshClientFor(t)
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "nuke: SSH connect failed for %s: %v\n", t.Host, err)
|
|
return
|
|
}
|
|
defer client.Close()
|
|
unit := "gapmind-" + t.Service + ".service"
|
|
remoteDir := t.RemoteDir
|
|
if remoteDir == "" {
|
|
remoteDir = ".local/share/gapmind"
|
|
}
|
|
remoteBinary := strings.TrimSuffix(remoteDir, "/") + "/gapmind-" + t.Service
|
|
|
|
|
|
script := fmt.Sprintf(
|
|
`set +e; runtime=/run/user/$(id -u); export XDG_RUNTIME_DIR=$runtime DBUS_SESSION_BUS_ADDRESS=unix:path=$runtime/bus; `+
|
|
`systemctl --user stop %[1]q >/dev/null 2>&1; systemctl --user disable %[1]q >/dev/null 2>&1; `+
|
|
`rm -f .config/systemd/user/%[1]q; systemctl --user daemon-reload >/dev/null 2>&1; `+
|
|
`rm -f %[2]q %[2]q.upload; rm -rf %[3]q; rm -rf .local/share/gapmind/%[4]q*; echo NUKED`,
|
|
unit, remoteBinary, strings.TrimSuffix(remoteDir, "/"), t.Service,
|
|
)
|
|
out, err := client.Run("ssh", client.Remote(), "sh", "-lc", script)
|
|
fmt.Printf("nuke: %s@%s %s\n", t.Service, t.Host, strings.TrimSpace(string(out)))
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "nuke: remote wipe had errors on %s: %v\n", t.Host, err)
|
|
}
|
|
}
|
|
|
|
func nukeLocal() {
|
|
|
|
paths := []string{"gapmind.sqlite", "gapmind.sqlite-wal", "gapmind.sqlite-shm", "gapmind.sqlite-journal", "dist", "clients/typescript/gapmind.ts", "clients/rust/src/lib.rs"}
|
|
for _, p := range paths {
|
|
if err := os.RemoveAll(p); err != nil {
|
|
fmt.Fprintf(os.Stderr, "nuke: could not remove %s: %v\n", p, err)
|
|
continue
|
|
}
|
|
fmt.Printf("nuke: removed %s\n", p)
|
|
}
|
|
}
|
|
|
|
func deploy(target Target) {
|
|
if target.Host == "" || target.Service == "" || target.Binary == "" {
|
|
fail(fmt.Errorf("each target needs host, service, and binary"))
|
|
}
|
|
if target.Token == "" && target.Service != "dht" {
|
|
fail(fmt.Errorf("target %s has no shared peer token", target.Host))
|
|
}
|
|
if target.Service != "replica" && target.Service != "pathfinder" && target.Service != "dht" {
|
|
fail(fmt.Errorf("unknown service %q", target.Service))
|
|
}
|
|
remoteDir := target.RemoteDir
|
|
if remoteDir == "" {
|
|
remoteDir = ".local/share/gapmind"
|
|
}
|
|
remoteBinary := strings.TrimSuffix(remoteDir, "/") + "/gapmind-" + target.Service
|
|
execStart := remoteBinary
|
|
if target.Service == "dht" {
|
|
port := target.DHTListenPort
|
|
if port == 0 {
|
|
port = 6881
|
|
}
|
|
passive := true
|
|
if target.DHTPassive != nil {
|
|
passive = *target.DHTPassive
|
|
}
|
|
interval := target.RSSInterval
|
|
if interval == "" {
|
|
interval = "10m"
|
|
}
|
|
execStart = fmt.Sprintf("%s --listen :%d --passive=%t --interval %s", remoteBinary, port, passive, interval)
|
|
}
|
|
if !strings.HasPrefix(execStart, "/") {
|
|
execStart = "%h/" + strings.TrimPrefix(execStart, "./")
|
|
}
|
|
remote := target.Host
|
|
if target.User != "" {
|
|
remote = target.User + "@" + target.Host
|
|
}
|
|
environment := map[string]string{"CENTRAL_URL": target.CentralURL, "PEER_API_TOKEN": target.Token, "PEER_PUBLIC_URL": target.PublicURL}
|
|
if target.Service == "pathfinder" {
|
|
environment["NODE_LATITUDE"] = target.Latitude
|
|
environment["NODE_LONGITUDE"] = target.Longitude
|
|
}
|
|
if target.Service == "dht" {
|
|
port := target.DHTListenPort
|
|
if port == 0 {
|
|
port = 6881
|
|
}
|
|
environment["DHT_LISTEN"] = fmt.Sprintf(":%d", port)
|
|
if target.Trackers != "" {
|
|
environment["TRACKERS"] = target.Trackers
|
|
}
|
|
if target.RSSFeeds != "" {
|
|
environment["RSS_FEEDS"] = target.RSSFeeds
|
|
}
|
|
if target.DHTAPIToken != "" {
|
|
environment["DHT_API_TOKEN"] = target.DHTAPIToken
|
|
}
|
|
}
|
|
password := target.SSHPassword
|
|
if password == "" {
|
|
password = target.Token
|
|
}
|
|
unit := service.UserUnit{Description: "Gapmind " + target.Service, ExecStart: execStart, Environment: environment}
|
|
client, err := service.NewSSHClient(remote, password)
|
|
if err != nil {
|
|
fail(err)
|
|
}
|
|
defer client.Close()
|
|
if target.TimeoutSec > 0 {
|
|
client.SetTimeout(time.Duration(target.TimeoutSec) * time.Second)
|
|
}
|
|
fmt.Printf("Deploying %s to %s\n", target.Service, remote)
|
|
if err := service.CopyAndInstall(client, target.Binary, remoteBinary, "gapmind-"+target.Service+".service", unit); err != nil {
|
|
fail(err)
|
|
}
|
|
}
|
|
|
|
func cmdSync(targets []Target, restart bool) {
|
|
if len(targets) == 0 {
|
|
fmt.Fprintln(os.Stderr, "no targets to sync")
|
|
return
|
|
}
|
|
client := &http.Client{Timeout: 20 * time.Second}
|
|
|
|
var updates []peerUpdate
|
|
for _, t := range targets {
|
|
if t.Service != "pathfinder" && t.Service != "dht" {
|
|
continue
|
|
}
|
|
meta := map[string]interface{}{}
|
|
if t.Latitude != "" {
|
|
meta["latitude"] = t.Latitude
|
|
}
|
|
if t.Longitude != "" {
|
|
meta["longitude"] = t.Longitude
|
|
}
|
|
if t.Service == "pathfinder" {
|
|
meta["transport"] = "outbound-relay"
|
|
}
|
|
if t.Service == "dht" {
|
|
port := t.DHTListenPort
|
|
if port == 0 {
|
|
port = 6881
|
|
}
|
|
meta["dhtListenPort"] = port
|
|
meta["dhtPassive"] = true
|
|
if t.DHTPassive != nil {
|
|
meta["dhtPassive"] = *t.DHTPassive
|
|
}
|
|
}
|
|
meta["assembled"] = true
|
|
meta["assembledAt"] = time.Now().UTC().Format(time.RFC3339)
|
|
updates = append(updates, peerUpdate{
|
|
NodeId: t.resolvedNodeId(),
|
|
Metadata: meta,
|
|
Kind: t.Service,
|
|
})
|
|
}
|
|
if len(updates) == 0 {
|
|
fmt.Println("nothing to sync")
|
|
return
|
|
}
|
|
for _, t := range targets {
|
|
if t.CentralURL != "" {
|
|
syncDB(client, t.CentralURL, t.Token, updates)
|
|
break
|
|
}
|
|
}
|
|
|
|
if restart {
|
|
for _, t := range targets {
|
|
restartRemote(t)
|
|
}
|
|
}
|
|
}
|
|
|
|
func syncDB(client *http.Client, centralURL, token string, updates []peerUpdate) {
|
|
body, _ := json.Marshal(map[string]interface{}{"updates": updates})
|
|
req, _ := http.NewRequest("POST", strings.TrimRight(centralURL, "/")+"/api/peers/metadata", bytes.NewReader(body))
|
|
req.Header.Set("Content-Type", "application/json")
|
|
if token != "" {
|
|
req.Header.Set("Authorization", "Bearer "+token)
|
|
}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "sync: central API request failed: %v\n", err)
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
|
if resp.StatusCode >= 300 {
|
|
fmt.Fprintf(os.Stderr, "sync: central API returned HTTP %d: %s\n", resp.StatusCode, string(respBody))
|
|
return
|
|
}
|
|
var result struct {
|
|
Applied int `json:"applied"`
|
|
Rejected []struct {
|
|
NodeId string `json:"nodeId"`
|
|
Error string `json:"error"`
|
|
} `json:"rejected"`
|
|
}
|
|
if err := json.Unmarshal(respBody, &result); err != nil {
|
|
fmt.Fprintf(os.Stderr, "sync: could not parse response: %v\n", err)
|
|
return
|
|
}
|
|
fmt.Printf("sync: %d peers updated, %d rejected\n", result.Applied, len(result.Rejected))
|
|
for _, r := range result.Rejected {
|
|
fmt.Fprintf(os.Stderr, " rejected %s: %s\n", r.NodeId, r.Error)
|
|
}
|
|
}
|
|
|
|
func RestartUnit(client *service.SSHClient, unitName string) error {
|
|
if _, err := client.Run("sudo systemctl daemon-reload"); err != nil {
|
|
return fmt.Errorf("daemon-reload failed: %w", err)
|
|
}
|
|
if _, err := client.Run("sudo systemctl restart " + unitName); err != nil {
|
|
return fmt.Errorf("restart failed: %w", err)
|
|
}
|
|
if _, err := client.Run("sudo systemctl enable " + unitName); err != nil {
|
|
return fmt.Errorf("enable failed: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func restartRemote(target Target) {
|
|
remote := target.Host
|
|
if target.User != "" {
|
|
remote = target.User + "@" + target.Host
|
|
}
|
|
password := target.SSHPassword
|
|
if password == "" {
|
|
password = target.Token
|
|
}
|
|
client, err := service.NewSSHClient(remote, password)
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "restart: SSH connect failed for %s: %v\n", target.Host, err)
|
|
return
|
|
}
|
|
defer client.Close()
|
|
if target.TimeoutSec > 0 {
|
|
client.SetTimeout(time.Duration(target.TimeoutSec) * time.Second)
|
|
}
|
|
serviceName := "gapmind-" + target.Service + ".service"
|
|
fmt.Printf("restart: restarting %s on %s\n", serviceName, target.Host)
|
|
if err := RestartUnit(client, serviceName); err != nil {
|
|
fmt.Fprintf(os.Stderr, "restart: %s failed on %s: %v\n", serviceName, target.Host, err)
|
|
}
|
|
}
|
|
|
|
func fail(err error) { fmt.Fprintln(os.Stderr, "orchestra:", err); os.Exit(1) }
|