rsp2k/headplane
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: handle logging from the agent

Browse Source
This commit is contained in:
Aarnav Tale 2025-08-20 13:58:19 -04:00
parent bcd87453bd
commit 8fc657f86a
No known key found for this signature in database
10 changed files with 63 additions and 235 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
app/server/hp-agent.ts
View File

@ -1,8 +1,8 @@
import { ChildProcessWithoutNullStreams, spawn } from 'node:child_process';
import EventEmitter from 'node:events';
import { constants, access, mkdir, open } from 'node:fs/promises';
import { access, constants, mkdir, open } from 'node:fs/promises';
import { getegid, geteuid } from 'node:process';
import { Interface, createInterface } from 'node:readline';
import { createInterface, Interface } from 'node:readline';
import { inArray } from 'drizzle-orm';
import { LibSQLDatabase } from 'drizzle-orm/libsql/driver-core';
import { HostInfo } from '~/types';
@ -240,7 +240,6 @@ class HeadplaneAgent extends EventEmitter<AgentEvents> {
gid: getegid?.() ?? undefined,
env: {
HOME: process.env.HOME,
HEADPLANE_EMBEDDED: 'true',
HEADPLANE_AGENT_WORK_DIR: this.options.work_dir,
HEADPLANE_AGENT_DEBUG: log.debugEnabled ? 'true' : 'false',
HEADPLANE_AGENT_HOSTNAME: this.options.host_name,
@ -396,5 +395,26 @@ class HeadplaneAgent extends EventEmitter<AgentEvents> {
this.emit('error', new Error(error));
return;
}
if (line.startsWith('LOG')) {
const logSnippet = line.slice(4).trim();
const [level, ...messageParts] = logSnippet.split(' ');
const message = messageParts.join(' ');
switch (level) {
case 'INFO':
log.info('agent', message);
break;
case 'WARN':
log.warn('agent', message);
break;
case 'ERROR':
log.error('agent', message);
break;
default:
log.debug('agent', message);
}
return;
}
}
}

43
cmd/hp_agent/hp_agent.go
View File

@ -1,11 +1,13 @@
package main
import (
"bufio"
"context"
"fmt"
"os"
"strings"
_ "github.com/joho/godotenv/autoload"
"github.com/tale/headplane/internal/config"
"github.com/tale/headplane/internal/hpagent"
"github.com/tale/headplane/internal/tsnet"
"github.com/tale/headplane/internal/util"
)
@ -24,15 +26,36 @@ func main() {
log.SetDebug(cfg.Debug)
agent := tsnet.NewAgent(cfg)
agent.Connect()
defer agent.Shutdown()
log.Msg(&Register{
Type: "register",
ID: agent.ID,
})
scanner := bufio.NewScanner(os.Stdin)
for scanner.Scan() {
line := scanner.Bytes()
directive := strings.TrimSpace(string(line))
log.Debug("Received directive: %s", directive)
fmt.Println("READY")
hpagent.FollowMaster(agent)
switch directive {
case "START":
agent.Connect()
fmt.Printf("READY %s\n", agent.ID)
case "SHUTDOWN":
agent.Shutdown()
os.Exit(0)
case "PING":
fmt.Printf("PONG %s\n", agent.ID)
case "REFRESH":
err := agent.DispatchHostInfo(context.Background())
if err != nil {
fmt.Printf("ERROR %s\n", err)
}
}
}
if err := scanner.Err(); err != nil {
fmt.Printf("ERROR %s\n", err)
os.Exit(1)
}
}

1
go.mod
View File

@ -3,7 +3,6 @@ module github.com/tale/headplane
go 1.24.4
require (
github.com/joho/godotenv v1.5.1
go4.org/mem v0.0.0-20240501181205-ae6ca9944745
golang.org/x/crypto v0.41.0
tailscale.com v1.86.4

2
go.sum
View File

@ -99,8 +99,6 @@ github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9Y
github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8=
github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
github.com/jsimonetti/rtnetlink v1.4.0 h1:Z1BF0fRgcETPEa0Kt0MRk3yV5+kF1FWTni6KUFKrq2I=
github.com/jsimonetti/rtnetlink v1.4.0/go.mod h1:5W1jDvWdnthFJ7fxYX1GMK07BUpI4oskfOqvPteYS6E=
github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=

45
internal/config.go
View File

@ -1,45 +0,0 @@
package config
import "os"
// Config represents the configuration for the agent.
type Config struct {
Debug bool
Hostname string
TSControlURL string
TSAuthKey string
WorkDir string
}
const (
DebugEnv = "HEADPLANE_AGENT_DEBUG"
HostnameEnv = "HEADPLANE_AGENT_HOSTNAME"
TSControlURLEnv = "HEADPLANE_AGENT_TS_SERVER"
TSAuthKeyEnv = "HEADPLANE_AGENT_TS_AUTHKEY"
WorkDirEnv = "HEADPLANE_AGENT_WORK_DIR"
)
// Load reads the agent configuration from environment variables.
func Load() (*Config, error) {
c := &Config{
Debug: false,
Hostname: os.Getenv(HostnameEnv),
TSControlURL: os.Getenv(TSControlURLEnv),
TSAuthKey: os.Getenv(TSAuthKeyEnv),
WorkDir: os.Getenv(WorkDirEnv),
}
if os.Getenv(DebugEnv) == "true" {
c.Debug = true
}
if err := validateRequired(c); err != nil {
return nil, err
}
if err := validateTSReady(c); err != nil {
return nil, err
}
return c, nil
}

55
internal/hpagent/handler.go
View File

@ -1,55 +0,0 @@
package hpagent
import (
"bufio"
"context"
"fmt"
"os"
"github.com/tale/headplane/internal/tsnet"
"github.com/tale/headplane/internal/util"
)
// Starts listening for messages from stdin
func FollowMaster(agent *tsnet.TSAgent) {
log := util.GetLogger()
scanner := bufio.NewScanner(os.Stdin)
log.Info("Listening for messages from Headplane master on stdin")
for scanner.Scan() {
line := scanner.Bytes()
directive := string(line)
log.Debug("Received directive from master: %s", directive)
switch directive {
case "SHUTDOWN":
log.Debug("Received SHUTDOWN directive from master, shutting down agent")
agent.Shutdown()
return
case "START":
log.Debug("Received START directive from master, starting agent")
// TODO: Start the agent here instead of in main
fmt.Println("READY " + agent.ID)
continue
case "PING":
log.Debug("Received PING directive from master, responding with PONG")
fmt.Println("PONG " + agent.ID)
continue
case "REFRESH":
log.Debug("Received REFRESH directive from master, refreshing status for all nodes")
err := agent.DispatchHostInfo(context.Background())
if err != nil {
log.Error("Error refreshing host info: %s", err)
fmt.Println("ERR " + err.Error())
}
}
}
if err := scanner.Err(); err != nil {
log.Fatal("Error reading from stdin: %s", err)
}
}

48
internal/preflight.go
View File

@ -1,48 +0,0 @@
package config
import (
"fmt"
"net/http"
"strings"
)
// Checks to make sure all required environment variables are set
func validateRequired(config *Config) error {
if config.Hostname == "" {
return fmt.Errorf("%s is required", HostnameEnv)
}
if config.TSControlURL == "" {
return fmt.Errorf("%s is required", TSControlURLEnv)
}
if config.TSAuthKey == "" {
return fmt.Errorf("%s is required", TSAuthKeyEnv)
}
if config.WorkDir == "" {
return fmt.Errorf("%s is required", WorkDirEnv)
}
return nil
}
// Pings the Tailscale control server to make sure it's up and running
func validateTSReady(config *Config) error {
testURL := config.TSControlURL
if strings.HasSuffix(testURL, "/") {
testURL = testURL[:len(testURL)-1]
}
testURL = fmt.Sprintf("%s/health", testURL)
resp, err := http.Get(testURL)
if err != nil {
return fmt.Errorf("Failed to connect to TS control server: %s", err)
}
if resp.StatusCode != 200 {
return fmt.Errorf("Failed to connect to TS control server: %s", resp.Status)
}
return nil
}

74
internal/util/logger.go
View File

@ -4,19 +4,16 @@ import (
"encoding/json"
"fmt"
"os"
"strings"
"sync"
"time"
)
type LogLevel string
const (
LevelInfo LogLevel = "info"
LevelDebug LogLevel = "debug"
LevelError LogLevel = "error"
LevelFatal LogLevel = "fatal"
LevelMsg LogLevel = "msg"
LevelInfo LogLevel = "INFO"
LevelDebug LogLevel = "DEBUG"
LevelError LogLevel = "ERROR"
LevelFatal LogLevel = "FATAL"
)
type LogMessage struct {
@ -61,19 +58,8 @@ func (l *Logger) SetDebug(enabled bool) {
func (l *Logger) log(level LogLevel, format string, v ...any) {
msg := fmt.Sprintf(format, v...)
timestamp := time.Now().Format(time.RFC3339)
// Manually construct compact JSON line for performance
line := `{"Level":"` + string(level) +
`","Time":"` + timestamp +
`","Message":"` + escapeString(msg) + `"}` + "\n"
if level == LevelError || level == LevelFatal {
os.Stderr.WriteString(line)
}
// Always write to stdout but also write to stderr for errors
os.Stdout.WriteString(line)
fmt.Printf("LOG %s %s\n", level, msg)
if level == LevelFatal {
os.Exit(1)
}
@ -88,53 +74,3 @@ func (l *Logger) Debug(format string, v ...any) {
func (l *Logger) Info(format string, v ...any) { l.log(LevelInfo, format, v...) }
func (l *Logger) Error(format string, v ...any) { l.log(LevelError, format, v...) }
func (l *Logger) Fatal(format string, v ...any) { l.log(LevelFatal, format, v...) }
func (l *Logger) Msg(obj any) {
entry := l.pool.Get().(*LogMessage)
defer l.pool.Put(entry)
entry.Level = LevelMsg
entry.Time = time.Now().Format(time.RFC3339)
entry.Message = obj
// Because the encoder is tied to STDOUT we get a message
_ = l.encoder.Encode(entry)
// Reset the entry for reuse
entry.Level = ""
entry.Time = ""
entry.Message = nil
}
func escapeString(s string) string {
var b strings.Builder
b.Grow(len(s) + 16) // pre-grow to reduce reallocs
for i := 0; i < len(s); i++ {
c := s[i]
switch c {
case '"':
b.WriteString(`\"`)
case '\\':
b.WriteString(`\\`)
case '\b':
b.WriteString(`\b`)
case '\f':
b.WriteString(`\f`)
case '\n':
b.WriteString(`\n`)
case '\r':
b.WriteString(`\r`)
case '\t':
b.WriteString(`\t`)
default:
if c < 0x20 {
// Control characters like 0x01, 0x07 (bell), etc.
fmt.Fprintf(&b, `\u%04x`, c)
} else {
b.WriteByte(c)
}
}
}
return b.String()
}

2
nix/agent.nix
View File

@ -3,7 +3,7 @@ buildGoModule {
pname = "hp_agent";
version = (builtins.fromJSON (builtins.readFile ../package.json)).version;
src = ../.;
vendorHash = "sha256-AyJx1os4auTmHvz+KLY+RGyysa3LWi8r2+QLxofPUH4=";
vendorHash = "sha256-cPE8cnfTdzi6hAmSXujKmfd5ezivc3sQ6DKOZubCpYI=";
ldflags = ["-s" "-w"];
env.CGO_ENABLED = 0;
}

2
nix/ssh-wasm.nix
View File

@ -14,7 +14,7 @@ in
version = (builtins.fromJSON (builtins.readFile ../package.json)).version;
src = ../.;
subPackages = ["cmd/hp_ssh"];
vendorHash = "sha256-AyJx1os4auTmHvz+KLY+RGyysa3LWi8r2+QLxofPUH4=";
vendorHash = "sha256-cPE8cnfTdzi6hAmSXujKmfd5ezivc3sQ6DKOZubCpYI=";
env.CGO_ENABLED = 0;
nativeBuildInputs = [go];