cli.go 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620
  1. package raft
  2. import (
  3. "bufio"
  4. "bytes"
  5. "fmt"
  6. "net"
  7. "os"
  8. "os/exec"
  9. "path/filepath"
  10. "regexp"
  11. "runtime"
  12. "sort"
  13. "strconv"
  14. "strings"
  15. "sync"
  16. "text/tabwriter"
  17. "time"
  18. )
  19. const (
  20. ColorReset = "\033[0m"
  21. ColorDim = "\033[90m" // Dark Gray
  22. ColorRed = "\033[31m"
  23. ColorGreen = "\033[32m"
  24. ColorYellow = "\033[33m"
  25. ColorBlue = "\033[34m"
  26. ColorCyan = "\033[36m"
  27. )
  28. // CLICommand defines a CLI command handler
  29. type CLICommand struct {
  30. Name string
  31. Description string
  32. Handler func(parts []string, server *KVServer)
  33. }
  34. // CLI manages the command line interface
  35. type CLI struct {
  36. server *KVServer
  37. commands map[string]CLICommand
  38. mu sync.RWMutex
  39. token string
  40. }
  41. // NewCLI creates a new CLI instance
  42. func NewCLI(server *KVServer) *CLI {
  43. cli := &CLI{
  44. server: server,
  45. commands: make(map[string]CLICommand),
  46. }
  47. cli.registerDefaultCommands()
  48. return cli
  49. }
  50. // Helper to calculate visible length of string ignoring ANSI codes
  51. var ansiRegex = regexp.MustCompile(`\x1b\[[0-9;]*m`)
  52. func visibleLen(s string) int {
  53. clean := ansiRegex.ReplaceAllString(s, "")
  54. return len([]rune(clean))
  55. }
  56. func printBoxed(content string) {
  57. lines := strings.Split(strings.TrimSpace(content), "\n")
  58. maxWidth := 0
  59. for _, line := range lines {
  60. l := visibleLen(line)
  61. if l > maxWidth {
  62. maxWidth = l
  63. }
  64. }
  65. // Min width to look decent
  66. if maxWidth < 20 {
  67. maxWidth = 20
  68. }
  69. // Add padding
  70. contentWidth := maxWidth + 2
  71. fmt.Println() // Start new line before box
  72. // Top Border
  73. fmt.Printf("%s╭%s╮%s\n", ColorDim, strings.Repeat("─", contentWidth), ColorReset)
  74. // Content
  75. for _, line := range lines {
  76. visLen := visibleLen(line)
  77. padding := contentWidth - visLen - 1 // -1 for the space after │
  78. // Ensure padding is not negative
  79. if padding < 0 { padding = 0 }
  80. fmt.Printf("%s│%s %s%s%s│%s\n", ColorDim, ColorReset, line, strings.Repeat(" ", padding), ColorDim, ColorReset)
  81. }
  82. // Bottom Border
  83. fmt.Printf("%s╰%s╯%s\n", ColorDim, strings.Repeat("─", contentWidth), ColorReset)
  84. fmt.Println() // End new line after box
  85. }
  86. func formatBytes(bytes int64) string {
  87. const unit = 1024
  88. if bytes < unit {
  89. return fmt.Sprintf("%d B", bytes)
  90. }
  91. div, exp := int64(unit), 0
  92. for n := bytes / unit; n >= unit; n /= unit {
  93. div *= unit
  94. exp++
  95. }
  96. return fmt.Sprintf("%.1f %cB", float64(bytes)/float64(div), "KMGTPE"[exp])
  97. }
  98. // scanCRLF is a split function for a Scanner that returns each line of
  99. // text, stripped of any trailing end-of-line marker. The returned line may
  100. // be empty. The end-of-line marker is one optional carriage return followed
  101. // by one mandatory newline, OR just a carriage return.
  102. func scanCRLF(data []byte, atEOF bool) (advance int, token []byte, err error) {
  103. if atEOF && len(data) == 0 {
  104. return 0, nil, nil
  105. }
  106. if i := bytes.IndexByte(data, '\n'); i >= 0 {
  107. // We have a full newline-terminated line.
  108. return i + 1, dropCR(data[0:i]), nil
  109. }
  110. if i := bytes.IndexByte(data, '\r'); i >= 0 {
  111. // We have a CR-terminated line.
  112. return i + 1, data[0:i], nil
  113. }
  114. // If we're at EOF, we have a final, non-terminated line. Return it.
  115. if atEOF {
  116. return len(data), dropCR(data), nil
  117. }
  118. // Request more data.
  119. return 0, nil, nil
  120. }
  121. // dropCR drops a terminal \r from the data.
  122. func dropCR(data []byte) []byte {
  123. if len(data) > 0 && data[len(data)-1] == '\r' {
  124. return data[0 : len(data)-1]
  125. }
  126. return data
  127. }
  128. // Start starts the command line interface
  129. func (c *CLI) Start() {
  130. // Attempt to fix terminal settings if they are messed up (e.g. missing ICRNL)
  131. // This helps when Enter sends \r but the terminal driver expects \n
  132. cmd := exec.Command("stty", "sane")
  133. cmd.Stdin = os.Stdin
  134. cmd.Stdout = os.Stdout
  135. _ = cmd.Run()
  136. // Boxed startup message
  137. var sb strings.Builder
  138. sb.WriteString(fmt.Sprintf("Node %s%s%s CLI Started\n", ColorGreen, c.server.Raft.nodeID, ColorReset))
  139. sb.WriteString("Type 'help' for commands.")
  140. printBoxed(sb.String())
  141. // Initialize history handling
  142. historyPath := filepath.Join(os.TempDir(), fmt.Sprintf(".raft_history_%s", c.server.Raft.nodeID))
  143. historyFile, err := os.OpenFile(historyPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
  144. if err != nil {
  145. // Just warn, don't fail
  146. fmt.Printf("%sWarning: Could not open history file: %v%s\n", ColorYellow, err, ColorReset)
  147. }
  148. defer func() {
  149. if historyFile != nil {
  150. historyFile.Close()
  151. }
  152. }()
  153. // State Monitor Loop
  154. go func() {
  155. var lastState string
  156. var lastTerm uint64
  157. stats := c.server.GetStats()
  158. lastState = stats.State
  159. lastTerm = stats.Term
  160. ticker := time.NewTicker(100 * time.Millisecond)
  161. defer ticker.Stop()
  162. for {
  163. select {
  164. case <-c.server.stopCh:
  165. return
  166. case <-ticker.C:
  167. stats := c.server.GetStats()
  168. if stats.State != lastState || stats.Term != lastTerm {
  169. msg := fmt.Sprintf("%s[State Change] %s (Term %d) -> %s (Term %d)%s",
  170. ColorYellow, lastState, lastTerm, stats.State, stats.Term, ColorReset)
  171. fmt.Printf("\n%s\n> ", msg)
  172. lastState = stats.State
  173. lastTerm = stats.Term
  174. }
  175. }
  176. }
  177. }()
  178. scanner := bufio.NewScanner(os.Stdin)
  179. scanner.Split(scanCRLF)
  180. fmt.Print("> ")
  181. for scanner.Scan() {
  182. text := strings.TrimSpace(scanner.Text())
  183. if text == "" {
  184. fmt.Print("> ")
  185. continue
  186. }
  187. // Save command to history
  188. if historyFile != nil {
  189. if _, err := historyFile.WriteString(text + "\n"); err != nil {
  190. // Ignore write error
  191. }
  192. }
  193. parts := strings.Fields(text)
  194. cmdName := strings.ToLower(parts[0])
  195. c.mu.RLock()
  196. cmd, exists := c.commands[cmdName]
  197. c.mu.RUnlock()
  198. if exists {
  199. cmd.Handler(parts, c.server)
  200. } else {
  201. // Check aliases
  202. switch cmdName {
  203. case "binlog", "db", "stats":
  204. c.mu.RLock()
  205. if infoCmd, ok := c.commands["info"]; ok {
  206. infoCmd.Handler(parts, c.server)
  207. }
  208. c.mu.RUnlock()
  209. case "history":
  210. c.printHistory(historyPath)
  211. default:
  212. printBoxed("Unknown command. Type 'help'.")
  213. }
  214. }
  215. fmt.Print("> ")
  216. }
  217. }
  218. // printHistory prints the command history
  219. func (c *CLI) printHistory(path string) {
  220. content, err := os.ReadFile(path)
  221. if err != nil {
  222. printBoxed("No history found.")
  223. return
  224. }
  225. lines := strings.Split(string(content), "\n")
  226. // Show last 20 lines
  227. start := 0
  228. if len(lines) > 20 {
  229. start = len(lines) - 20
  230. }
  231. var sb strings.Builder
  232. sb.WriteString(fmt.Sprintf("%sCommand History (Last 20):%s\n", ColorCyan, ColorReset))
  233. for i := start; i < len(lines); i++ {
  234. line := strings.TrimSpace(lines[i])
  235. if line != "" {
  236. sb.WriteString(fmt.Sprintf("%d: %s\n", i+1, line))
  237. }
  238. }
  239. printBoxed(sb.String())
  240. }
  241. // RegisterCommand registers a new CLI command
  242. func (c *CLI) RegisterCommand(name, description string, handler func(parts []string, server *KVServer)) {
  243. c.mu.Lock()
  244. defer c.mu.Unlock()
  245. c.commands[strings.ToLower(name)] = CLICommand{
  246. Name: name,
  247. Description: description,
  248. Handler: handler,
  249. }
  250. }
  251. func (c *CLI) registerDefaultCommands() {
  252. c.RegisterCommand("set", "Set value (set <key> <val>)", func(parts []string, server *KVServer) {
  253. if len(parts) != 3 {
  254. printBoxed("Usage: set <key> <value>")
  255. return
  256. }
  257. key, val := parts[1], parts[2]
  258. // Use Authenticated Set
  259. c.mu.RLock()
  260. token := c.token
  261. c.mu.RUnlock()
  262. if err := server.SetAuthenticated(key, val, token); err != nil {
  263. printBoxed(fmt.Sprintf("%sError:%s %v", ColorRed, ColorReset, err))
  264. } else {
  265. printBoxed(fmt.Sprintf("%sOK%s", ColorGreen, ColorReset))
  266. }
  267. })
  268. c.RegisterCommand("get", "Get value (get <key>)", func(parts []string, server *KVServer) {
  269. if len(parts) != 2 {
  270. printBoxed("Usage: get <key>")
  271. return
  272. }
  273. key := parts[1]
  274. c.mu.RLock()
  275. token := c.token
  276. c.mu.RUnlock()
  277. if val, ok, err := server.GetLinearAuthenticated(key, token); err != nil {
  278. printBoxed(fmt.Sprintf("%sError:%s %v", ColorRed, ColorReset, err))
  279. } else if !ok {
  280. printBoxed(fmt.Sprintf("%sNot Found%s", ColorYellow, ColorReset))
  281. } else {
  282. printBoxed(fmt.Sprintf("%s%s%s = %s%s%s", ColorCyan, key, ColorReset, ColorYellow, val, ColorReset))
  283. }
  284. })
  285. c.RegisterCommand("del", "Delete key (del <key>)", func(parts []string, server *KVServer) {
  286. if len(parts) != 2 {
  287. printBoxed("Usage: del <key>")
  288. return
  289. }
  290. key := parts[1]
  291. c.mu.RLock()
  292. token := c.token
  293. c.mu.RUnlock()
  294. if err := server.DelAuthenticated(key, token); err != nil {
  295. printBoxed(fmt.Sprintf("%sError:%s %v", ColorRed, ColorReset, err))
  296. } else {
  297. printBoxed(fmt.Sprintf("%sDeleted%s", ColorGreen, ColorReset))
  298. }
  299. })
  300. c.RegisterCommand("login", "Login to system (login <user> <pass> [code])", func(parts []string, server *KVServer) {
  301. if len(parts) < 3 {
  302. printBoxed("Usage: login <username> <password> [mfa_code]")
  303. return
  304. }
  305. username := parts[1]
  306. password := parts[2]
  307. code := ""
  308. if len(parts) > 3 {
  309. code = parts[3]
  310. }
  311. token, err := server.AuthManager.Login(username, password, code, "cli")
  312. if err != nil {
  313. printBoxed(fmt.Sprintf("%sLogin Failed:%s %v", ColorRed, ColorReset, err))
  314. } else {
  315. c.mu.Lock()
  316. c.token = token
  317. c.mu.Unlock()
  318. printBoxed(fmt.Sprintf("%sLogin Success! Token saved.%s", ColorGreen, ColorReset))
  319. }
  320. })
  321. c.RegisterCommand("logout", "Logout from system", func(parts []string, server *KVServer) {
  322. c.mu.Lock()
  323. c.token = ""
  324. c.mu.Unlock()
  325. printBoxed("Logged out.")
  326. })
  327. c.RegisterCommand("auth-init", "Initialize Auth System (auth-init <root_pass>)", func(parts []string, server *KVServer) {
  328. if len(parts) != 2 {
  329. printBoxed("Usage: auth-init <root_password>")
  330. return
  331. }
  332. // 1. Create Root User
  333. if err := server.AuthManager.CreateRootUser(parts[1]); err != nil {
  334. printBoxed(fmt.Sprintf("Failed to create root user: %v", err))
  335. return
  336. }
  337. // 2. Enable Auth
  338. if err := server.AuthManager.EnableAuth(); err != nil {
  339. printBoxed(fmt.Sprintf("Failed to enable auth: %v", err))
  340. return
  341. }
  342. printBoxed("Auth Initialized. Please login as 'root'.")
  343. })
  344. c.RegisterCommand("whoami", "Show current user info", func(parts []string, server *KVServer) {
  345. c.mu.RLock()
  346. token := c.token
  347. c.mu.RUnlock()
  348. if token == "" {
  349. printBoxed("Not logged in.")
  350. return
  351. }
  352. // Decode token or check session
  353. // Accessing AuthManager internal session map is protected.
  354. // We can add a helper in AuthManager to get session info or just show token.
  355. printBoxed(fmt.Sprintf("Logged in (Token: %s...)", token[:8]))
  356. })
  357. c.RegisterCommand("join", "Add node to cluster (join <id> <addr>)", func(parts []string, server *KVServer) {
  358. if len(parts) != 3 {
  359. printBoxed("Usage: join <nodeID> <address>")
  360. return
  361. }
  362. nodeID := parts[1]
  363. addr := parts[2]
  364. if err := server.Join(nodeID, addr); err != nil {
  365. printBoxed(fmt.Sprintf("%sError joining node:%s %v", ColorRed, ColorReset, err))
  366. } else {
  367. printBoxed(fmt.Sprintf("%sJoin request sent for node %s%s", ColorGreen, nodeID, ColorReset))
  368. }
  369. })
  370. c.RegisterCommand("leave", "Remove node from cluster (leave <id>)", func(parts []string, server *KVServer) {
  371. if len(parts) != 2 {
  372. printBoxed("Usage: leave <nodeID>")
  373. return
  374. }
  375. nodeID := parts[1]
  376. if err := server.Leave(nodeID); err != nil {
  377. printBoxed(fmt.Sprintf("%sError leaving node:%s %v", ColorRed, ColorReset, err))
  378. } else {
  379. printBoxed(fmt.Sprintf("%sLeave request sent for node %s%s", ColorGreen, nodeID, ColorReset))
  380. }
  381. })
  382. c.RegisterCommand("search", "Search keys (search <pat> [lim] [off])", func(parts []string, server *KVServer) {
  383. if len(parts) < 2 {
  384. printBoxed("Usage: search <pattern> [limit] [offset]")
  385. return
  386. }
  387. pattern := parts[1]
  388. limit := 20 // Default limit
  389. offset := 0
  390. if len(parts) >= 3 {
  391. if l, err := strconv.Atoi(parts[2]); err == nil {
  392. limit = l
  393. }
  394. }
  395. if len(parts) >= 4 {
  396. if o, err := strconv.Atoi(parts[3]); err == nil {
  397. offset = o
  398. }
  399. }
  400. // Construct SQL for DB Engine
  401. sql := fmt.Sprintf("key like \"%s\" LIMIT %d OFFSET %d", pattern, limit, offset)
  402. results, err := server.DB.Query(sql)
  403. if err != nil {
  404. printBoxed(fmt.Sprintf("%sError:%s %v", ColorRed, ColorReset, err))
  405. return
  406. }
  407. // Format Table to Buffer
  408. var buf bytes.Buffer
  409. w := tabwriter.NewWriter(&buf, 0, 0, 2, ' ', 0)
  410. fmt.Fprintln(w, "Key\tValue\tCommitIndex")
  411. fmt.Fprintln(w, "---\t-----\t-----------")
  412. for _, r := range results {
  413. fmt.Fprintf(w, "%s\t%s\t%d\n", r.Key, r.Value, r.CommitIndex)
  414. }
  415. w.Flush()
  416. summary := fmt.Sprintf("\n%sFound %d records (showing max %d)%s", ColorDim, len(results), limit, ColorReset)
  417. printBoxed(buf.String() + summary)
  418. })
  419. c.RegisterCommand("count", "Count keys (count <pat>)", func(parts []string, server *KVServer) {
  420. if len(parts) < 2 {
  421. printBoxed("Usage: count <pattern> (e.g. 'count *', 'count user.*')")
  422. return
  423. }
  424. pattern := parts[1]
  425. sql := ""
  426. if pattern == "*" {
  427. sql = "*"
  428. } else {
  429. sql = fmt.Sprintf("key like \"%s\"", pattern)
  430. }
  431. start := time.Now()
  432. count, err := server.DB.Count(sql)
  433. duration := time.Since(start)
  434. if err != nil {
  435. printBoxed(fmt.Sprintf("%sError:%s %v", ColorRed, ColorReset, err))
  436. } else {
  437. printBoxed(fmt.Sprintf("%sCount: %d%s (took %v)", ColorGreen, count, ColorReset, duration))
  438. }
  439. })
  440. c.RegisterCommand("info", "Show all system stats", func(parts []string, server *KVServer) {
  441. // 1. Gather all data
  442. var m runtime.MemStats
  443. runtime.ReadMemStats(&m)
  444. dbSize := server.GetDBSize()
  445. logSize := server.GetLogSize()
  446. stats := server.GetStats()
  447. health := server.HealthCheck()
  448. dbApplied := server.DB.GetLastAppliedIndex()
  449. // 2. Build Output
  450. var sb strings.Builder
  451. // Section: Node Status
  452. sb.WriteString(fmt.Sprintf("%sNode Status:%s\n", ColorCyan, ColorReset))
  453. sb.WriteString(fmt.Sprintf(" ID: %s%s%s\n", ColorGreen, health.NodeID, ColorReset))
  454. sb.WriteString(fmt.Sprintf(" State: %s\n", health.State))
  455. sb.WriteString(fmt.Sprintf(" Term: %d\n", health.Term))
  456. sb.WriteString(fmt.Sprintf(" Leader: %s\n", health.LeaderID))
  457. sb.WriteString(fmt.Sprintf(" Healthy: %v\n", health.IsHealthy))
  458. sb.WriteString("\n")
  459. // Section: Storage Stats
  460. sb.WriteString(fmt.Sprintf("%sStorage & Memory:%s\n", ColorCyan, ColorReset))
  461. sb.WriteString(fmt.Sprintf(" DB Size: %s\n", formatBytes(dbSize)))
  462. sb.WriteString(fmt.Sprintf(" Raft Log: %s\n", formatBytes(logSize)))
  463. sb.WriteString(fmt.Sprintf(" Mem Alloc: %s\n", formatBytes(int64(m.Alloc))))
  464. sb.WriteString(fmt.Sprintf(" Mem Sys: %s\n", formatBytes(int64(m.Sys))))
  465. sb.WriteString(fmt.Sprintf(" NumGC: %d\n", m.NumGC))
  466. sb.WriteString("\n")
  467. // Section: Raft & DB Indices
  468. sb.WriteString(fmt.Sprintf("%sConsensus Indices:%s\n", ColorCyan, ColorReset))
  469. sb.WriteString(fmt.Sprintf(" Commit Index: %d\n", stats.CommitIndex))
  470. sb.WriteString(fmt.Sprintf(" Applied Index:%d\n", stats.LastApplied))
  471. sb.WriteString(fmt.Sprintf(" Last Log Idx: %d\n", stats.LastLogIndex))
  472. sb.WriteString(fmt.Sprintf(" DB Applied: %d\n", dbApplied))
  473. sb.WriteString("\n")
  474. // Section: Cluster Members
  475. sb.WriteString(fmt.Sprintf("%sCluster Members (%d):%s", ColorCyan, stats.ClusterSize, ColorReset))
  476. for id, addr := range stats.ClusterNodes {
  477. // Determine status color
  478. statusColor := ColorDim
  479. nodeID := server.Raft.nodeID // Assuming local nodeID is accessible
  480. if id == nodeID {
  481. statusColor = ColorCyan // Self
  482. } else {
  483. // Quick check
  484. conn, err := net.DialTimeout("tcp", addr, 50*time.Millisecond)
  485. if err == nil {
  486. statusColor = ColorGreen
  487. conn.Close()
  488. }
  489. }
  490. // Marker for Leader
  491. marker := ""
  492. if id == health.LeaderID {
  493. marker = fmt.Sprintf(" %s(Leader)%s", ColorYellow, ColorReset)
  494. }
  495. sb.WriteString(fmt.Sprintf("\n - %s%s%s: %s%s", statusColor, id, ColorReset, addr, marker))
  496. }
  497. printBoxed(sb.String())
  498. })
  499. c.RegisterCommand("quit", "Gracefully shutdown the node", func(parts []string, server *KVServer) {
  500. printBoxed("Shutting down node...")
  501. if err := server.Stop(); err != nil {
  502. printBoxed(fmt.Sprintf("%sError stopping server:%s %v", ColorRed, ColorReset, err))
  503. } else {
  504. printBoxed(fmt.Sprintf("%sNode stopped successfully.%s", ColorGreen, ColorReset))
  505. }
  506. os.Exit(0)
  507. })
  508. c.RegisterCommand("clear", "Clear screen", func(parts []string, server *KVServer) {
  509. fmt.Print("\033[H\033[2J")
  510. })
  511. c.RegisterCommand("help", "Show commands", func(parts []string, server *KVServer) {
  512. var sb strings.Builder
  513. sb.WriteString(fmt.Sprintf("%sCommands:%s\n", ColorGreen, ColorReset))
  514. // Sort commands
  515. c.mu.RLock()
  516. var cmds []CLICommand
  517. for _, cmd := range c.commands {
  518. cmds = append(cmds, cmd)
  519. }
  520. c.mu.RUnlock()
  521. sort.Slice(cmds, func(i, j int) bool {
  522. return cmds[i].Name < cmds[j].Name
  523. })
  524. // Find max length for padding
  525. maxNameLen := 0
  526. for _, cmd := range cmds {
  527. if len(cmd.Name) > maxNameLen {
  528. maxNameLen = len(cmd.Name)
  529. }
  530. }
  531. for _, cmd := range cmds {
  532. padding := strings.Repeat(" ", maxNameLen-len(cmd.Name))
  533. sb.WriteString(fmt.Sprintf(" %s%s%s%s %s%s%s\n", ColorGreen, cmd.Name, ColorReset, padding, ColorDim, cmd.Description, ColorReset))
  534. }
  535. printBoxed(sb.String())
  536. })
  537. }