diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 8990247..d70c6f6 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -12,6 +12,7 @@ import ( "time" "github.com/flatrun/agent/internal/api" + "github.com/flatrun/agent/internal/observ" "github.com/flatrun/agent/internal/watcher" "github.com/flatrun/agent/pkg/config" "github.com/flatrun/agent/pkg/updater" @@ -106,7 +107,17 @@ func newRootCmd() *cobra.Command { }, } - root.AddCommand(serve, setup, update, versionCmd) + observPlugin := &cobra.Command{ + Use: "observ-plugin", + Short: "Run the built-in observability plugin (launched by the agent)", + Hidden: true, + SilenceUsage: true, + RunE: func(cmd *cobra.Command, args []string) error { + return observ.RunPlugin() + }, + } + + root.AddCommand(serve, setup, update, versionCmd, observPlugin) return root } diff --git a/cmd/observ-plugin/main.go b/cmd/observ-plugin/main.go new file mode 100644 index 0000000..b02978e --- /dev/null +++ b/cmd/observ-plugin/main.go @@ -0,0 +1,12 @@ +// Command observ-plugin is the FlatRun observability app as a standalone binary. The agent +// normally runs this logic by re-execing itself; this command allows building the plugin on +// its own for development or as an external plugin. +// +// Build: go build -o plugin ./cmd/observ-plugin +package main + +import "github.com/flatrun/agent/internal/observ" + +func main() { + _ = observ.RunPlugin() +} diff --git a/examples/plugins/hello/main.go b/examples/plugins/hello/main.go index 8fdf385..def0615 100644 --- a/examples/plugins/hello/main.go +++ b/examples/plugins/hello/main.go @@ -20,10 +20,25 @@ func main() { _, _ = w.Write([]byte(`{"message":"hello from the flatrun plugin"}`)) }) + echo := pluginsdk.Tool{ + Spec: pluginapi.ToolSpec{ + Name: "echo", + Description: "Echo back the given text.", + Parameters: map[string]any{ + "type": "object", + "properties": map[string]any{"text": map[string]any{"type": "string"}}, + }, + }, + Run: func(args map[string]any) (string, error) { + text, _ := args["text"].(string) + return "echo: " + text, nil + }, + } + _ = pluginsdk.Serve(pluginapi.Info{ Name: "hello", Version: "0.1.0", DisplayName: "Hello Plugin", Description: "A sample out-of-process plugin.", - }, mux) + }, mux, echo) } diff --git a/go.mod b/go.mod index f0b8711..6e7a6f2 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ require ( github.com/digitalocean/godo v1.171.0 github.com/distribution/reference v0.6.0 github.com/docker/docker v28.5.2+incompatible - github.com/fsnotify/fsnotify v1.9.0 + github.com/fsnotify/fsnotify v1.10.1 github.com/gin-contrib/cors v1.7.6 github.com/gin-gonic/gin v1.10.1 github.com/go-sql-driver/mysql v1.9.3 @@ -21,12 +21,13 @@ require ( github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 github.com/lib/pq v1.10.9 github.com/moby/moby/client v0.2.2 + github.com/nicholas-fedor/shoutrrr v0.16.1 github.com/robfig/cron/v3 v3.0.1 github.com/spf13/cobra v1.10.2 github.com/testcontainers/testcontainers-go v0.41.0 github.com/testcontainers/testcontainers-go/modules/compose v0.41.0 - golang.org/x/crypto v0.48.0 - golang.org/x/oauth2 v0.35.0 + golang.org/x/crypto v0.51.0 + golang.org/x/oauth2 v0.36.0 gopkg.in/yaml.v3 v3.0.1 modernc.org/sqlite v1.47.0 ) @@ -77,6 +78,7 @@ require ( github.com/docker/go-units v0.5.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/ebitengine/purego v0.10.0 // indirect + github.com/eclipse/paho.golang v0.23.0 // indirect github.com/eiannone/keyboard v0.0.0-20220611211555-0d226195f203 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsevents v0.2.0 // indirect @@ -113,7 +115,8 @@ require ( github.com/leodido/go-urn v1.4.0 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect - github.com/mattn/go-isatty v0.0.20 // indirect + github.com/mattn/go-colorable v0.1.15 // indirect + github.com/mattn/go-isatty v0.0.22 // indirect github.com/mattn/go-runewidth v0.0.16 // indirect github.com/mattn/go-shellwords v1.0.12 // indirect github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect @@ -138,7 +141,7 @@ require ( github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pelletier/go-toml v1.9.5 // indirect - github.com/pelletier/go-toml/v2 v2.2.4 // indirect + github.com/pelletier/go-toml/v2 v2.3.1 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect @@ -185,11 +188,11 @@ require ( go.yaml.in/yaml/v3 v3.0.4 // indirect go.yaml.in/yaml/v4 v4.0.0-rc.4 // indirect golang.org/x/arch v0.18.0 // indirect - golang.org/x/net v0.50.0 // indirect - golang.org/x/sync v0.19.0 // indirect - golang.org/x/sys v0.42.0 // indirect - golang.org/x/term v0.40.0 // indirect - golang.org/x/text v0.34.0 // indirect + golang.org/x/net v0.55.0 // indirect + golang.org/x/sync v0.21.0 // indirect + golang.org/x/sys v0.46.0 // indirect + golang.org/x/term v0.44.0 // indirect + golang.org/x/text v0.38.0 // indirect golang.org/x/time v0.14.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect diff --git a/go.sum b/go.sum index 36e28b9..ef647cd 100644 --- a/go.sum +++ b/go.sum @@ -10,6 +10,8 @@ github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEK github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= github.com/DefangLabs/secret-detector v0.0.0-20250403165618-22662109213e h1:rd4bOvKmDIx0WeTv9Qz+hghsgyjikFiPrseXHlKepO0= github.com/DefangLabs/secret-detector v0.0.0-20250403165618-22662109213e/go.mod h1:blbwPQh4DTlCZEfk1BLU4oMIhLda2U+A840Uag9DsZw= +github.com/Masterminds/semver/v3 v3.5.0 h1:kQceYJfbupGfZOKZQg0kou0DgAKhzDg2NZPAwZ/2OOE= +github.com/Masterminds/semver/v3 v3.5.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/Microsoft/hcsshim v0.14.0-rc.1 h1:qAPXKwGOkVn8LlqgBN8GS0bxZ83hOJpcjxzmlQKxKsQ= @@ -152,6 +154,8 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ISU= github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= +github.com/eclipse/paho.golang v0.23.0 h1:KHgl2wz6EJo7cMBmkuhpt7C576vP+kpPv7jjvSyR6Mk= +github.com/eclipse/paho.golang v0.23.0/go.mod h1:nQRhTkoZv8EAiNs5UU0/WdQIx2NrnWUpL9nsGJTQN04= github.com/eiannone/keyboard v0.0.0-20220611211555-0d226195f203 h1:XBBHcIb256gUJtLmY22n99HaZTz+r2Z51xUPi01m3wg= github.com/eiannone/keyboard v0.0.0-20220611211555-0d226195f203/go.mod h1:E1jcSv8FaEny+OP/5k9UxZVw9YFWGj7eI4KR/iOBqCg= github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM= @@ -160,8 +164,8 @@ github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/fsnotify/fsevents v0.2.0 h1:BRlvlqjvNTfogHfeBOFvSC9N0Ddy+wzQCQukyoD7o/c= github.com/fsnotify/fsevents v0.2.0/go.mod h1:B3eEk39i4hz8y1zaWS/wPrAP4O6wkIl7HQwKBr1qH/w= -github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= -github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= +github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho= +github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo= github.com/fvbommel/sortorder v1.1.0 h1:fUmoe+HLsBTctBDoaBwpQo5N+nrCp8g/BjKb/6ZQmYw= github.com/fvbommel/sortorder v1.1.0/go.mod h1:uk88iVf1ovNn1iLfgUVU2F9o5eO30ui720w+kxuqRs0= github.com/gabriel-vasile/mimetype v1.4.9 h1:5k+WDwEsD9eTLL8Tz3L0VnmVh9QxGjRmjBvAG7U/oYY= @@ -231,6 +235,8 @@ github.com/go-playground/validator/v10 v10.26.0 h1:SP05Nqhjcvz81uJaRfEV0YBSSSGMc github.com/go-playground/validator/v10 v10.26.0/go.mod h1:I5QpIEbmr8On7W0TktmJAumgzX4CA1XNl4ZmDuVHKKo= github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo= github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU= +github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI= +github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4= @@ -256,8 +262,8 @@ github.com/google/go-containerregistry v0.20.7/go.mod h1:Lx5LCZQjLH1QBaMPeGwsME9 github.com/google/go-querystring v1.1.0 h1:AnCroh3fv4ZBgVIf1Iwtovgjaw/GiKJo8M8yD/fhyJ8= github.com/google/go-querystring v1.1.0/go.mod h1:Kcdr2DB4koayq7X8pmAG4sNG59So17icRSOU623lUBU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= -github.com/google/pprof v0.0.0-20250820193118-f64d9cf942d6 h1:EEHtgt9IwisQ2AZ4pIsMjahcegHh6rmhqxzIRQIyepY= -github.com/google/pprof v0.0.0-20250820193118-f64d9cf942d6/go.mod h1:I6V7YzU0XDpsHqbsyrghnFZLO1gwK6NPTNvmetQIk9U= +github.com/google/pprof v0.0.0-20260604005048-7023385849c0 h1:h1QTMDl6q9wDvDCJVpKQSjgleGFYnd2fOxmg2K+6BGE= +github.com/google/pprof v0.0.0-20260604005048-7023385849c0/go.mod h1:MxpfABSjhmINe3F1It9d+8exIHFvUqtLIRCdOGNXqiI= github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4= github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= @@ -291,6 +297,8 @@ github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/inhies/go-bytesize v0.0.0-20220417184213-4913239db9cf h1:FtEj8sfIcaaBfAKrE1Cwb61YDtYq9JxChK1c7AKce7s= github.com/inhies/go-bytesize v0.0.0-20220417184213-4913239db9cf/go.mod h1:yrqSXGoD/4EKfF26AOGzscPOgTTJcyAwM2rpixWT+t4= +github.com/jarcoal/httpmock v1.4.1 h1:0Ju+VCFuARfFlhVXFc2HxlcQkfB+Xq12/EotHko+x2A= +github.com/jarcoal/httpmock v1.4.1/go.mod h1:ftW1xULwo+j0R0JJkJIIi7UKigZUXCLLanykgjwBXL0= github.com/jonboulle/clockwork v0.5.0 h1:Hyh9A8u51kptdkR+cqRpT1EebBwTn1oK9YfGYbdFz6I= github.com/jonboulle/clockwork v0.5.0/go.mod h1:3mZlmanh0g2NDKO5TWZVJAfofYk64M7XN3SzBPjZF60= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= @@ -315,10 +323,10 @@ github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= -github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= -github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= -github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= -github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-colorable v0.1.15 h1:+u9SLTRGnXv73cEsnsmoZBom+dMU88B2M0aDcWy0/jY= +github.com/mattn/go-colorable v0.1.15/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= +github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4= +github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4= github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc= github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w= github.com/mattn/go-shellwords v1.0.12 h1:M2zGm7EW6UQJvDeQxo4T51eKPurbeFbe8WtebGE2xrk= @@ -369,8 +377,14 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/nicholas-fedor/shoutrrr v0.16.1 h1:MOzl3U6zprA45lOTKaTbQW/9wzkR7SC36IFrTr5xfkU= +github.com/nicholas-fedor/shoutrrr v0.16.1/go.mod h1:lii4gQKKSV2c32b+D93ysEdKpEzVla6HTbyOajpNDpo= github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4= github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= +github.com/onsi/ginkgo/v2 v2.29.0 h1:rfh+ZFjgJhYWRoIqVf3Uwx/W20yLrcrE2h2GmYVRaag= +github.com/onsi/ginkgo/v2 v2.29.0/go.mod h1:+aXOY+vzZ5mu2iI2HpTZUPmM//oQfsNFX6gU9kNcA44= +github.com/onsi/gomega v1.41.0 h1:OwKp4pXNgVxf6sCplzYo794OFNuoL2q2SBMU5NSWOjA= +github.com/onsi/gomega v1.41.0/go.mod h1:M/Uqpu/8qTjtzCLUA2zJHX9Iilrau25x1PdoSRbWh5A= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= @@ -381,8 +395,8 @@ github.com/opencontainers/selinux v1.13.1 h1:A8nNeceYngH9Ow++M+VVEwJVpdFmrlxsN22 github.com/opencontainers/selinux v1.13.1/go.mod h1:S10WXZ/osk2kWOYKy1x2f/eXF5ZHJoUs8UU/2caNRbg= github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3ve8= github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c= -github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= -github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/pelletier/go-toml/v2 v2.3.1 h1:MYEvvGnQjeNkRF1qUuGolNtNExTDwct51yp7olPtrEc= +github.com/pelletier/go-toml/v2 v2.3.1/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo= @@ -444,8 +458,8 @@ github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3A github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= -github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= -github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= @@ -541,25 +555,25 @@ golang.org/x/arch v0.18.0/go.mod h1:bdwinDaKcfZUGpH09BB7ZmOfhalA8lQdzl62l8gGWsk= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= -golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= +golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI= +golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8= -golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= +golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= +golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= -golang.org/x/net v0.50.0 h1:ucWh9eiCGyDR3vtzso0WMQinm2Dnt8cFMuQa9K33J60= -golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM= -golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ= -golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= +golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= +golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= -golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= +golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -569,23 +583,22 @@ golang.org/x/sys v0.0.0-20210331175145-43e1dd70ce54/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= -golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/term v0.40.0 h1:36e4zGLqU4yhjlmxEaagx2KuYbJq3EwY8K943ZsHcvg= -golang.org/x/term v0.40.0/go.mod h1:w2P8uVp06p2iyKKuvXIm7N/y0UCRt3UfJTfZ7oOpglM= +golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= +golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.44.0 h1:0rLvDRCtNj0gZkyIXhCyOb2OAzEhLVqc4B+hrsBhrmc= +golang.org/x/term v0.44.0/go.mod h1:7ze4MdzUzLXpSAoFP1H0bOI9aXDqveSvatT5vKcFh2Y= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= -golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE= +golang.org/x/text v0.38.0/go.mod h1:YXZt3QhHUKYT53r2lLKFIVi6Ao1jdzrTR/KQ09qyxF4= golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= -golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= -golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= +golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8= +golang.org/x/tools v0.45.0/go.mod h1:LuUGqqaXcXMEFEruIVJVm5mgDD8vww/z/SR1gQ4uE/0= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/internal/api/ai_tools.go b/internal/api/ai_tools.go index 833aaa6..6d51e45 100644 --- a/internal/api/ai_tools.go +++ b/internal/api/ai_tools.go @@ -14,6 +14,7 @@ import ( "github.com/flatrun/agent/internal/ai" "github.com/flatrun/agent/internal/auth" "github.com/flatrun/agent/internal/setup" + "github.com/flatrun/agent/pkg/pluginapi" "github.com/gin-gonic/gin" ) @@ -348,6 +349,24 @@ func (s *Server) aiToolRegistry() map[string]aiTool { // aiToolSpecs returns the tool schemas to advertise to the model, in a // stable order. +// pluginToolPrefix namespaces plugin-provided tools so they never collide with built-ins and +// can be routed back to the owning plugin on dispatch: plugin____. +const pluginToolPrefix = "plugin__" + +func pluginToolName(plugin, tool string) string { + return pluginToolPrefix + plugin + "__" + tool +} + +// parsePluginToolName splits a namespaced tool name back into plugin and tool. +func parsePluginToolName(name string) (plugin, tool string, ok bool) { + rest, found := strings.CutPrefix(name, pluginToolPrefix) + if !found { + return "", "", false + } + plugin, tool, found = strings.Cut(rest, "__") + return plugin, tool, found +} + func (s *Server) aiToolSpecs() []ai.Tool { registry := s.aiToolRegistry() names := make([]string, 0, len(registry)) @@ -359,26 +378,98 @@ func (s *Server) aiToolSpecs() []ai.Tool { for _, name := range names { specs = append(specs, registry[name].Spec) } + + // Append tools contributed by running plugins. + for _, info := range s.pluginInfos() { + for _, t := range info.Tools { + specs = append(specs, ai.Tool{ + Name: pluginToolName(info.Name, t.Name), + Description: t.Description, + Parameters: t.Parameters, + }) + } + } return specs } +// pluginInfos returns running plugin infos, tolerating a nil host (as in unit tests). +func (s *Server) pluginInfos() []pluginapi.Info { + if s.pluginHost == nil { + return nil + } + return s.pluginHost.Infos() +} + +// pluginToolMutates reports whether a namespaced plugin tool changes state, so the caller +// applies the write-access gate. +func (s *Server) pluginToolMutates(plugin, tool string) bool { + for _, info := range s.pluginInfos() { + if info.Name != plugin { + continue + } + for _, t := range info.Tools { + if t.Name == tool { + return t.Mutates + } + } + } + return false +} + // runAITool executes one tool call, returning the textual result the // model reads. Errors are returned as content (prefixed) so the model // can recover rather than the loop aborting. func (s *Server) runAITool(c *gin.Context, boundDeployment string, call ai.ToolCall) string { - tool, ok := s.aiToolRegistry()[call.Name] - if !ok { - return "Error: unknown tool " + call.Name - } var args map[string]interface{} if strings.TrimSpace(call.Arguments) != "" { if err := json.Unmarshal([]byte(call.Arguments), &args); err != nil { return "Error: could not parse tool arguments: " + err.Error() } } + + // Plugin-provided tools are dispatched to the owning plugin over its socket. + if plugin, tool, ok := parsePluginToolName(call.Name); ok { + if s.pluginToolMutates(plugin, tool) { + // A state-changing tool requires write access to a deployment, and the plugin is + // told which one so it can scope the mutation to that deployment rather than act + // on an arbitrary resource the caller named. + dep, err := s.toolAllowedDeploymentWrite(c, boundDeployment, args) + if err != nil { + return "Error: " + err.Error() + } + if args == nil { + args = map[string]interface{}{} + } + args["_deployment"] = dep + } + result, err := s.pluginHost.ExecTool(plugin, tool, args) + if err != nil { + return "Error: " + err.Error() + } + return truncateToolOutput(result) + } + + tool, ok := s.aiToolRegistry()[call.Name] + if !ok { + return "Error: unknown tool " + call.Name + } result, err := tool.Run(s, c, boundDeployment, args) if err != nil { return "Error: " + err.Error() } return result } + +// toolAllowedDeploymentWrite resolves the target deployment and verifies the actor may write +// to it, gating a mutating plugin tool. +func (s *Server) toolAllowedDeploymentWrite(c *gin.Context, boundDeployment string, args map[string]interface{}) (string, error) { + name := toolDeployment(boundDeployment, args) + if name == "" { + return "", fmt.Errorf("no deployment specified") + } + actor := auth.GetActorFromContext(c) + if actor != nil && actor.Role != auth.RoleAdmin && !actor.CanAccessDeployment(name, auth.AccessLevelWrite) { + return "", fmt.Errorf("you do not have write access to deployment %q", name) + } + return name, nil +} diff --git a/internal/api/ai_tools_test.go b/internal/api/ai_tools_test.go new file mode 100644 index 0000000..b02b9db --- /dev/null +++ b/internal/api/ai_tools_test.go @@ -0,0 +1,20 @@ +package api + +import "testing" + +func TestPluginToolNameRoundTrip(t *testing.T) { + name := pluginToolName("observability", "get_deployment_health") + if name != "plugin__observability__get_deployment_health" { + t.Fatalf("unexpected namespaced name %q", name) + } + plugin, tool, ok := parsePluginToolName(name) + if !ok || plugin != "observability" || tool != "get_deployment_health" { + t.Errorf("parse = (%q, %q, %v), want observability/get_deployment_health/true", plugin, tool, ok) + } +} + +func TestParsePluginToolNameRejectsBuiltins(t *testing.T) { + if _, _, ok := parsePluginToolName("get_instance_info"); ok { + t.Error("a built-in tool name must not parse as a plugin tool") + } +} diff --git a/internal/api/server.go b/internal/api/server.go index 88539c4..8d47c6e 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -38,6 +38,7 @@ import ( "github.com/flatrun/agent/internal/files" "github.com/flatrun/agent/internal/infra" "github.com/flatrun/agent/internal/networks" + "github.com/flatrun/agent/internal/notify" "github.com/flatrun/agent/internal/plan" "github.com/flatrun/agent/internal/pluginhost" "github.com/flatrun/agent/internal/proxy" @@ -73,6 +74,8 @@ type Server struct { firewall *firewall.Plugin builtinDNS []plugins.Plugin pluginHost *pluginhost.Host + notify *notify.Service + pluginToken string authMiddleware *auth.Middleware authManager *auth.Manager proxyOrchestrator *proxy.Orchestrator @@ -177,12 +180,21 @@ func New(cfg *config.Config, configPath string) *Server { } agentURL := fmt.Sprintf("http://127.0.0.1:%d/api", cfg.API.Port) + // A per-run secret handed to plugins so they can call the internal emit endpoint (e.g. to + // raise a notification) without the full user auth flow. + pluginToken := randomToken() + notifyService := notify.NewService(cfg.DeploymentsPath) pluginHost := pluginhost.New( filepath.Join(cfg.DeploymentsPath, ".flatrun", "plugins"), filepath.Join(cfg.DeploymentsPath, ".flatrun", "run"), agentURL, - "", + pluginToken, ) + // Built-in apps ship inside the agent and run by re-execing it with a subcommand, so + // there is one binary to deploy and no per-arch plugin artifact. + if self, err := os.Executable(); err == nil { + pluginHost.Builtin("observability", self, "observ-plugin") + } authMiddleware := auth.NewMiddleware(&cfg.Auth) setupManager := setup.NewManager(cfg, configPath) @@ -300,6 +312,8 @@ func New(cfg *config.Config, configPath string) *Server { firewall: firewallPlugin, builtinDNS: builtinDNS, pluginHost: pluginHost, + notify: notifyService, + pluginToken: pluginToken, authMiddleware: authMiddleware, authManager: authManager, proxyOrchestrator: proxyOrchestrator, @@ -446,6 +460,9 @@ func (s *Server) setupRoutes() { protected.GET("/settings", s.authMiddleware.RequirePermission(auth.PermSettingsRead), s.getSettings) protected.PUT("/settings", s.authMiddleware.RequirePermission(auth.PermSettingsWrite), s.updateSettings) protected.PUT("/settings/security", s.authMiddleware.RequirePermission(auth.PermSettingsWrite), s.updateSecuritySettings) + protected.GET("/notifications/targets", s.authMiddleware.RequirePermission(auth.PermSettingsRead), s.getNotificationTargets) + protected.PUT("/notifications/targets", s.authMiddleware.RequirePermission(auth.PermSettingsWrite), s.updateNotificationTargets) + protected.POST("/notifications/test", s.authMiddleware.RequirePermission(auth.PermSettingsWrite), s.testNotification) protected.GET("/config", s.authMiddleware.RequirePermission(auth.PermConfigRead), s.listConfig) protected.GET("/config/*key", s.authMiddleware.RequirePermission(auth.PermConfigRead), s.getConfigKey) protected.PUT("/config/*key", s.authMiddleware.RequirePermission(auth.PermConfigWrite), s.updateConfigKey) @@ -741,6 +758,9 @@ func (s *Server) setupRoutes() { // Cluster exchange endpoint (no auth - uses invite token) api.POST("/cluster/exchange", s.clusterExchange) + // Plugin-emitted notifications (authenticated by the per-run plugin token). + api.POST("/internal/notify/emit", s.emitNotification) + // Ingest endpoints (no auth - called by nginx Lua) api.POST("/security/events/ingest", s.ingestSecurityEvent) api.POST("/traffic/ingest", s.ingestTrafficLog) @@ -2949,12 +2969,18 @@ func (s *Server) listPlugins(c *gin.Context) { pluginList := s.pluginRegistry.List() for _, info := range s.pluginHost.Infos() { + exts := make([]plugins.UIExtension, len(info.UIExtensions)) + for i, e := range info.UIExtensions { + exts[i] = plugins.UIExtension{Slot: e.Slot, Kind: e.Kind, Title: e.Title, Icon: e.Icon, Endpoint: e.Endpoint} + } pluginList = append(pluginList, plugins.PluginInfo{ Name: info.Name, Version: info.Version, DisplayName: info.DisplayName, Description: info.Description, Capabilities: info.Capabilities, + UIExtensions: exts, + ConfigSchema: info.ConfigSchema, Type: plugins.TypeIntegration, Enabled: true, }) @@ -2965,6 +2991,67 @@ func (s *Server) listPlugins(c *gin.Context) { }) } +func randomToken() string { + buf := make([]byte, 24) + if _, err := cryptoRand.Read(buf); err != nil { + // A predictable fallback would be a known plugin-auth secret; fail loudly at startup + // instead, since a system without a working CSPRNG cannot be secured anyway. + panic("randomToken: crypto/rand unavailable: " + err.Error()) + } + return hex.EncodeToString(buf) +} + +func (s *Server) getNotificationTargets(c *gin.Context) { + c.JSON(http.StatusOK, s.notify.Load()) +} + +func (s *Server) updateNotificationTargets(c *gin.Context) { + var cfg notify.Config + if err := c.ShouldBindJSON(&cfg); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) + return + } + if err := s.notify.Save(cfg); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, s.notify.Load()) +} + +func (s *Server) testNotification(c *gin.Context) { + var req struct { + URL string `json:"url"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request"}) + return + } + if err := s.notify.Test(req.URL); err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "delivery failed: " + err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"message": "test notification sent"}) +} + +// emitNotification is called by out-of-process plugins (authenticated by the per-run plugin +// token) to raise a notification, which the core routes to the configured targets. +func (s *Server) emitNotification(c *gin.Context) { + if s.pluginToken == "" || c.GetHeader("X-Plugin-Token") != s.pluginToken { + c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"}) + return + } + var req struct { + Title string `json:"title"` + Message string `json:"message"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request"}) + return + } + _ = s.notify.Notify(req.Title, req.Message) + c.JSON(http.StatusAccepted, gin.H{"status": "queued"}) +} + func marketplaceAPIBase() string { if v := os.Getenv("FLATRUN_MARKETPLACE_API"); v != "" { return v diff --git a/internal/notify/notify.go b/internal/notify/notify.go new file mode 100644 index 0000000..50d0cd7 --- /dev/null +++ b/internal/notify/notify.go @@ -0,0 +1,96 @@ +// Package notify is FlatRun's core notification service. It stores delivery targets (email, +// webhook, chat, ... as shoutrrr URLs), routes events to the enabled ones, and can send a +// test message. It is a core capability: plugins emit events and the core delivers them, so +// notification configuration lives in one place rather than per plugin. +package notify + +import ( + "fmt" + "os" + "path/filepath" + "sync" + + "github.com/nicholas-fedor/shoutrrr" + "gopkg.in/yaml.v3" +) + +// Target is one delivery destination. URL is a shoutrrr service URL, e.g. +// "smtp://user:pass@host:587/?from=x&to=y" or "generic+https://example.com/hook". +type Target struct { + ID string `yaml:"id" json:"id"` + Name string `yaml:"name" json:"name"` + URL string `yaml:"url" json:"url"` + Enabled bool `yaml:"enabled" json:"enabled"` +} + +// Config is the persisted notification settings. +type Config struct { + Targets []Target `yaml:"targets" json:"targets"` +} + +// Service loads/saves targets and delivers messages. +type Service struct { + path string + mu sync.RWMutex + send func(url, message string) error // overridable in tests +} + +func NewService(basePath string) *Service { + return &Service{ + path: filepath.Join(basePath, ".flatrun", "notifications.yml"), + send: shoutrrr.Send, + } +} + +func (s *Service) Load() Config { + s.mu.RLock() + defer s.mu.RUnlock() + var cfg Config + data, err := os.ReadFile(s.path) + if err != nil { + return cfg + } + _ = yaml.Unmarshal(data, &cfg) + return cfg +} + +func (s *Service) Save(cfg Config) error { + s.mu.Lock() + defer s.mu.Unlock() + if err := os.MkdirAll(filepath.Dir(s.path), 0755); err != nil { + return err + } + data, err := yaml.Marshal(cfg) + if err != nil { + return err + } + return os.WriteFile(s.path, data, 0644) +} + +// Test sends a message to a single URL, so an admin can verify a target before saving it. +func (s *Service) Test(url string) error { + if url == "" { + return fmt.Errorf("no target url") + } + return s.send(url, "FlatRun test notification: your target is configured correctly.") +} + +// Notify delivers title + message to every enabled target. It returns the first delivery +// error, if any, but attempts all targets. +func (s *Service) Notify(title, message string) error { + cfg := s.Load() + body := message + if title != "" { + body = title + "\n\n" + message + } + var firstErr error + for _, t := range cfg.Targets { + if !t.Enabled || t.URL == "" { + continue + } + if err := s.send(t.URL, body); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr +} diff --git a/internal/notify/notify_test.go b/internal/notify/notify_test.go new file mode 100644 index 0000000..38f0374 --- /dev/null +++ b/internal/notify/notify_test.go @@ -0,0 +1,82 @@ +package notify + +import ( + "errors" + "testing" +) + +func TestServiceRoundTripAndNotify(t *testing.T) { + s := NewService(t.TempDir()) + + var sent []string + s.send = func(url, msg string) error { sent = append(sent, url+"|"+msg); return nil } + + cfg := Config{Targets: []Target{ + {ID: "1", Name: "ops email", URL: "smtp://x", Enabled: true}, + {ID: "2", Name: "disabled", URL: "smtp://y", Enabled: false}, + {ID: "3", Name: "webhook", URL: "generic+https://h", Enabled: true}, + }} + if err := s.Save(cfg); err != nil { + t.Fatal(err) + } + if got := s.Load(); len(got.Targets) != 3 { + t.Fatalf("expected 3 targets persisted, got %d", len(got.Targets)) + } + + if err := s.Notify("Alert", "web is unhealthy"); err != nil { + t.Fatalf("Notify = %v", err) + } + // Only the two enabled targets receive the message. + if len(sent) != 2 { + t.Fatalf("expected 2 deliveries (enabled only), got %d: %v", len(sent), sent) + } + for _, s := range sent { + if want := "Alert\n\nweb is unhealthy"; !contains(s, want) { + t.Errorf("delivery missing title+body: %q", s) + } + } +} + +func TestServiceTest(t *testing.T) { + s := NewService(t.TempDir()) + called := "" + s.send = func(url, msg string) error { called = url; return nil } + if err := s.Test("smtp://x"); err != nil { + t.Fatal(err) + } + if called != "smtp://x" { + t.Errorf("Test sent to %q", called) + } + if err := s.Test(""); err == nil { + t.Error("Test with empty url should error") + } +} + +func TestNotifyReturnsFirstError(t *testing.T) { + s := NewService(t.TempDir()) + _ = s.Save(Config{Targets: []Target{ + {ID: "1", URL: "a", Enabled: true}, + {ID: "2", URL: "b", Enabled: true}, + }}) + s.send = func(url, _ string) error { + if url == "a" { + return errors.New("boom") + } + return nil + } + if err := s.Notify("t", "m"); err == nil { + t.Error("expected first delivery error surfaced") + } +} + +func contains(s, sub string) bool { + return len(s) >= len(sub) && (s == sub || indexOf(s, sub) >= 0) +} +func indexOf(s, sub string) int { + for i := 0; i+len(sub) <= len(s); i++ { + if s[i:i+len(sub)] == sub { + return i + } + } + return -1 +} diff --git a/internal/observ/api.go b/internal/observ/api.go new file mode 100644 index 0000000..518bbc7 --- /dev/null +++ b/internal/observ/api.go @@ -0,0 +1,183 @@ +package observ + +import ( + "encoding/json" + "net/http" + "time" +) + +// healthReporter is the slice of the health watcher the API needs; nil-safe so the handler +// can be used in tests without a watcher. +type healthReporter interface { + Snapshot() []ContainerHealth + Events() []RecoveryEvent +} + +// configAccess is the slice of the config store the API needs. +type configAccess interface { + Load() Config + Save(Config) error +} + +// Handler serves the store, health, and config over JSON for the native FlatRun UI. It is +// mounted by the plugin and reached through the agent's plugin proxy. health, cfg, and apply +// may be nil. apply is invoked with the saved config so a live watcher/collector picks up a +// change without a restart. +func Handler(store *Store, health healthReporter, cfg configAccess, apply func(Config)) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/metrics/latest", func(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, groupLatest(store.Latest())) + }) + mux.HandleFunc("/metrics/deployment", func(w http.ResponseWriter, r *http.Request) { + name := r.URL.Query().Get("name") + writeJSON(w, filterByDeployment(groupLatest(store.Latest()), name)) + }) + mux.HandleFunc("/metrics/timeseries", func(w http.ResponseWriter, r *http.Request) { + deployment := r.URL.Query().Get("deployment") + since := time.Now().Add(-15 * time.Minute) + if v := r.URL.Query().Get("since"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + since = time.Now().Add(-d) + } + } + writeJSON(w, TimeSeriesResponse{Deployment: deployment, Metrics: buildTimeSeries(store, deployment, since)}) + }) + mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { + if health == nil { + writeJSON(w, []ContainerHealth{}) + return + } + writeJSON(w, filterHealth(health.Snapshot(), r.URL.Query().Get("deployment"))) + }) + mux.HandleFunc("/config", func(w http.ResponseWriter, r *http.Request) { + if cfg == nil { + writeJSON(w, DefaultConfig()) + return + } + if r.Method == http.MethodPut { + var incoming Config + if err := json.NewDecoder(r.Body).Decode(&incoming); err != nil { + http.Error(w, "invalid config", http.StatusBadRequest) + return + } + if err := cfg.Save(incoming); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if apply != nil { + apply(cfg.Load()) + } + } + writeJSON(w, cfg.Load()) + }) + mux.HandleFunc("/health/events", func(w http.ResponseWriter, _ *http.Request) { + if health == nil { + writeJSON(w, []RecoveryEvent{}) + return + } + writeJSON(w, health.Events()) + }) + mux.HandleFunc("/metrics/series", func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + key := SeriesKey{ + Deployment: q.Get("deployment"), + Container: q.Get("container"), + Metric: q.Get("metric"), + } + since := time.Time{} + if v := q.Get("since"); v != "" { + if secs, err := time.ParseDuration(v); err == nil { + since = time.Now().Add(-secs) + } + } + writeJSON(w, map[string]any{ + "deployment": key.Deployment, + "container": key.Container, + "metric": key.Metric, + "samples": store.Range(key, since), + }) + }) + return mux +} + +type containerMetrics struct { + Container string `json:"container"` + Metrics map[string]float64 `json:"metrics"` + Updated time.Time `json:"updated"` +} + +type deploymentMetrics struct { + Deployment string `json:"deployment"` + Containers []containerMetrics `json:"containers"` +} + +// groupLatest folds the flat latest-per-series list into deployment -> container -> metrics, +// the shape the UI renders as cards. +func groupLatest(points []LatestPoint) []deploymentMetrics { + type ck struct{ dep, cont string } + order := []ck{} + byContainer := map[ck]*containerMetrics{} + + for _, p := range points { + k := ck{p.Deployment, p.Container} + cm := byContainer[k] + if cm == nil { + cm = &containerMetrics{Container: p.Container, Metrics: map[string]float64{}} + byContainer[k] = cm + order = append(order, k) + } + cm.Metrics[p.Metric] = p.Value + if p.Time.After(cm.Updated) { + cm.Updated = p.Time + } + } + + depOrder := []string{} + byDeployment := map[string]*deploymentMetrics{} + for _, k := range order { + dm := byDeployment[k.dep] + if dm == nil { + dm = &deploymentMetrics{Deployment: k.dep} + byDeployment[k.dep] = dm + depOrder = append(depOrder, k.dep) + } + dm.Containers = append(dm.Containers, *byContainer[k]) + } + + out := make([]deploymentMetrics, 0, len(depOrder)) + for _, d := range depOrder { + out = append(out, *byDeployment[d]) + } + return out +} + +func filterByDeployment(deps []deploymentMetrics, name string) []deploymentMetrics { + if name == "" { + return deps + } + out := make([]deploymentMetrics, 0, 1) + for _, d := range deps { + if d.Deployment == name { + out = append(out, d) + } + } + return out +} + +func filterHealth(states []ContainerHealth, deployment string) []ContainerHealth { + if deployment == "" { + return states + } + out := make([]ContainerHealth, 0, len(states)) + for _, s := range states { + if s.Deployment == deployment { + out = append(out, s) + } + } + return out +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} diff --git a/internal/observ/api_test.go b/internal/observ/api_test.go new file mode 100644 index 0000000..d425cf7 --- /dev/null +++ b/internal/observ/api_test.go @@ -0,0 +1,111 @@ +package observ + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestHandlerLatestGroupsByDeployment(t *testing.T) { + store := NewStore(10) + now := time.Unix(1_700_000_000, 0) + store.Record(ContainerSample{Deployment: "shop", Container: "shop-web", CPUPercent: 10, MemoryUsage: 200}, now) + store.Record(ContainerSample{Deployment: "shop", Container: "shop-db", CPUPercent: 4, MemoryUsage: 500}, now) + + rec := httptest.NewRecorder() + Handler(store, nil, nil, nil).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics/latest", nil)) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d", rec.Code) + } + var got []deploymentMetrics + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode: %v\n%s", err, rec.Body.String()) + } + if len(got) != 1 || got[0].Deployment != "shop" { + t.Fatalf("expected one deployment 'shop', got %+v", got) + } + if len(got[0].Containers) != 2 { + t.Fatalf("expected 2 containers, got %d", len(got[0].Containers)) + } + var web *containerMetrics + for i := range got[0].Containers { + if got[0].Containers[i].Container == "shop-web" { + web = &got[0].Containers[i] + } + } + if web == nil || web.Metrics[MetricCPUUsage] != 10 || web.Metrics[MetricMemoryUsage] != 200 { + t.Errorf("shop-web container = %+v", web) + } +} + +func TestHandlerConfigRoundTrip(t *testing.T) { + cfg := NewConfigStore(t.TempDir()) + var applied *Config + h := Handler(NewStore(10), nil, cfg, func(c Config) { applied = &c }) + + body := `{"sample_interval_seconds":10,"auto_restart":false,"restart_cooldown_seconds":300}` + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest(http.MethodPut, "/config", strings.NewReader(body))) + if rec.Code != http.StatusOK { + t.Fatalf("PUT /config status = %d", rec.Code) + } + if applied == nil { + t.Fatal("apply was not called on PUT /config") + } + if applied.AutoRestart != false { + t.Errorf("apply got stale config: %+v", *applied) + } + + rec = httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/config", nil)) + var got Config + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got.SampleIntervalSeconds != 10 || got.AutoRestart != false || got.RestartCooldownSeconds != 300 { + t.Errorf("config not persisted: %+v", got) + } +} + +func TestHandlerMetricsDeploymentFilters(t *testing.T) { + store := NewStore(10) + now := time.Unix(1_700_000_000, 0) + store.Record(ContainerSample{Deployment: "shop", Container: "shop-web", CPUPercent: 5}, now) + store.Record(ContainerSample{Deployment: "blog", Container: "blog-web", CPUPercent: 9}, now) + + rec := httptest.NewRecorder() + Handler(store, nil, nil, nil).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics/deployment?name=shop", nil)) + var got []deploymentMetrics + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if len(got) != 1 || got[0].Deployment != "shop" { + t.Errorf("deployment filter = %+v, want only shop", got) + } +} + +func TestHandlerSeriesReturnsSamples(t *testing.T) { + store := NewStore(10) + base := time.Now().Add(-time.Minute) + for i := 0; i < 3; i++ { + store.Record(ContainerSample{Deployment: "d", Container: "c", CPUPercent: float64(i)}, base.Add(time.Duration(i)*time.Second)) + } + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/metrics/series?deployment=d&container=c&metric="+MetricCPUUsage+"&since=5m", nil) + Handler(store, nil, nil, nil).ServeHTTP(rec, req) + + var got struct { + Samples []Sample `json:"samples"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode: %v", err) + } + if len(got.Samples) != 3 { + t.Errorf("expected 3 samples, got %d", len(got.Samples)) + } +} diff --git a/internal/observ/collector.go b/internal/observ/collector.go new file mode 100644 index 0000000..42b6c2f --- /dev/null +++ b/internal/observ/collector.go @@ -0,0 +1,153 @@ +package observ + +import ( + "context" + "encoding/json" + "os/exec" + "strconv" + "strings" + "time" +) + +// StatsSource returns the current per-container readings. It is injectable so the collector +// can be tested without Docker. +type StatsSource func() ([]ContainerSample, error) + +// Collector periodically reads a StatsSource and records the readings into a Store. +type Collector struct { + store *Store + source StatsSource + interval time.Duration + now func() time.Time +} + +func NewCollector(store *Store, source StatsSource, interval time.Duration) *Collector { + if interval <= 0 { + interval = 5 * time.Second + } + return &Collector{store: store, source: source, interval: interval, now: time.Now} +} + +// collectOnce reads the source once and records the readings. Exposed for tests. +func (c *Collector) collectOnce() error { + samples, err := c.source() + if err != nil { + return err + } + t := c.now() + for _, s := range samples { + c.store.Record(s, t) + } + return nil +} + +// Run collects on each tick until ctx is cancelled. A failing read is skipped, not fatal, +// so a transient Docker hiccup does not stop collection. +func (c *Collector) Run(ctx context.Context) { + ticker := time.NewTicker(c.interval) + defer ticker.Stop() + _ = c.collectOnce() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + _ = c.collectOnce() + } + } +} + +// DockerStatsSource reads a point-in-time snapshot via `docker stats`, tagging each +// container with its compose project (deployment) from container labels. +func DockerStatsSource() ([]ContainerSample, error) { + deployments := containerDeployments() + + out, err := exec.Command("docker", "stats", "--no-stream", "--format", "{{json .}}").Output() + if err != nil { + return nil, err + } + + var samples []ContainerSample + for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") { + if line == "" { + continue + } + var raw struct { + Container string `json:"Container"` + Name string `json:"Name"` + CPUPerc string `json:"CPUPerc"` + MemUsage string `json:"MemUsage"` + NetIO string `json:"NetIO"` + } + if err := json.Unmarshal([]byte(line), &raw); err != nil { + continue + } + rx, tx := splitPair(raw.NetIO) + used, limit := splitPair(raw.MemUsage) + deployment := deployments[raw.Container] + if deployment == "" { + deployment = deployments[raw.Name] + } + samples = append(samples, ContainerSample{ + Deployment: deployment, + Container: raw.Name, + CPUPercent: parsePercent(raw.CPUPerc), + MemoryUsage: parseBytes(used), + MemoryLimit: parseBytes(limit), + NetworkRx: parseBytes(rx), + NetworkTx: parseBytes(tx), + }) + } + return samples, nil +} + +func containerDeployments() map[string]string { + m := make(map[string]string) + out, err := exec.Command("docker", "ps", "-a", "--format", + "{{.ID}}|{{.Names}}|{{.Label \"com.docker.compose.project\"}}").Output() + if err != nil { + return m + } + for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") { + parts := strings.SplitN(line, "|", 3) + if len(parts) != 3 || parts[2] == "" { + continue + } + m[parts[0]] = parts[2] + m[parts[1]] = parts[2] + } + return m +} + +func splitPair(s string) (string, string) { + parts := strings.Split(s, " / ") + if len(parts) != 2 { + return "", "" + } + return strings.TrimSpace(parts[0]), strings.TrimSpace(parts[1]) +} + +func parsePercent(s string) float64 { + v, _ := strconv.ParseFloat(strings.TrimSuffix(s, "%"), 64) + return v +} + +func parseBytes(s string) uint64 { + s = strings.ToUpper(strings.TrimSpace(s)) + mult := uint64(1) + switch { + case strings.HasSuffix(s, "KIB") || strings.HasSuffix(s, "KB"): + mult = 1 << 10 + s = strings.TrimSuffix(strings.TrimSuffix(s, "KIB"), "KB") + case strings.HasSuffix(s, "MIB") || strings.HasSuffix(s, "MB"): + mult = 1 << 20 + s = strings.TrimSuffix(strings.TrimSuffix(s, "MIB"), "MB") + case strings.HasSuffix(s, "GIB") || strings.HasSuffix(s, "GB"): + mult = 1 << 30 + s = strings.TrimSuffix(strings.TrimSuffix(s, "GIB"), "GB") + case strings.HasSuffix(s, "B"): + s = strings.TrimSuffix(s, "B") + } + v, _ := strconv.ParseFloat(strings.TrimSpace(s), 64) + return uint64(v * float64(mult)) +} diff --git a/internal/observ/collector_test.go b/internal/observ/collector_test.go new file mode 100644 index 0000000..6a13966 --- /dev/null +++ b/internal/observ/collector_test.go @@ -0,0 +1,57 @@ +package observ + +import ( + "testing" + "time" +) + +func TestCollectorRecordsFromSource(t *testing.T) { + store := NewStore(10) + samples := []ContainerSample{ + {Deployment: "shop", Container: "shop-web", CPUPercent: 10, MemoryUsage: 200}, + {Deployment: "shop", Container: "shop-db", CPUPercent: 3, MemoryUsage: 500}, + } + c := NewCollector(store, func() ([]ContainerSample, error) { return samples, nil }, time.Second) + fixed := time.Unix(1_700_000_000, 0) + c.now = func() time.Time { return fixed } + + if err := c.collectOnce(); err != nil { + t.Fatalf("collectOnce: %v", err) + } + + web := store.Range(SeriesKey{Deployment: "shop", Container: "shop-web", Metric: MetricCPUUsage}, fixed) + if len(web) != 1 || web[0].Value != 10 { + t.Errorf("web cpu = %+v, want one sample of 10", web) + } + // Two containers x five semconv metrics. + if got := len(store.Series()); got != 10 { + t.Errorf("expected 10 series, got %d", got) + } +} + +func TestParseBytes(t *testing.T) { + cases := map[string]uint64{ + "0B": 0, + "512B": 512, + "1KiB": 1024, + "1.5MiB": uint64(1.5 * (1 << 20)), + "2GiB": 2 << 30, + "100MB": 100 << 20, + "": 0, + } + for in, want := range cases { + if got := parseBytes(in); got != want { + t.Errorf("parseBytes(%q) = %d, want %d", in, got, want) + } + } +} + +func TestSplitPair(t *testing.T) { + a, b := splitPair("1.2MiB / 512MiB") + if a != "1.2MiB" || b != "512MiB" { + t.Errorf("splitPair = %q,%q", a, b) + } + if x, y := splitPair("garbage"); x != "" || y != "" { + t.Errorf("splitPair(garbage) = %q,%q, want empty", x, y) + } +} diff --git a/internal/observ/config.go b/internal/observ/config.go new file mode 100644 index 0000000..b98edc6 --- /dev/null +++ b/internal/observ/config.go @@ -0,0 +1,79 @@ +package observ + +import ( + "os" + "path/filepath" + "sync" + "time" + + "gopkg.in/yaml.v3" +) + +// Config is the observability app's user-tunable settings, stored flat in +// .flatrun/observability.yml. +type Config struct { + SampleIntervalSeconds int `yaml:"sample_interval_seconds" json:"sample_interval_seconds"` + AutoRestart bool `yaml:"auto_restart" json:"auto_restart"` + RestartCooldownSeconds int `yaml:"restart_cooldown_seconds" json:"restart_cooldown_seconds"` +} + +// DefaultConfig returns the built-in defaults. +func DefaultConfig() Config { + return Config{SampleIntervalSeconds: 5, AutoRestart: true, RestartCooldownSeconds: 120} +} + +func (c Config) sampleInterval() time.Duration { + if c.SampleIntervalSeconds <= 0 { + return 5 * time.Second + } + return time.Duration(c.SampleIntervalSeconds) * time.Second +} + +func (c Config) restartCooldown() time.Duration { + if c.RestartCooldownSeconds <= 0 { + return 2 * time.Minute + } + return time.Duration(c.RestartCooldownSeconds) * time.Second +} + +// ConfigStore loads and saves the config from a flat file. +type ConfigStore struct { + path string + mu sync.RWMutex +} + +func NewConfigStore(basePath string) *ConfigStore { + return &ConfigStore{path: filepath.Join(basePath, ".flatrun", "observability.yml")} +} + +func (s *ConfigStore) Load() Config { + s.mu.RLock() + defer s.mu.RUnlock() + cfg := DefaultConfig() + data, err := os.ReadFile(s.path) + if err != nil { + return cfg + } + _ = yaml.Unmarshal(data, &cfg) + return cfg +} + +func (s *ConfigStore) Save(cfg Config) error { + s.mu.Lock() + defer s.mu.Unlock() + if err := os.MkdirAll(filepath.Dir(s.path), 0755); err != nil { + return err + } + data, err := yaml.Marshal(cfg) + if err != nil { + return err + } + return os.WriteFile(s.path, data, 0644) +} + +// ConfigSchema describes the config for the settings form the UI renders. +var ConfigSchema = map[string]any{ + "sample_interval_seconds": map[string]any{"type": "number", "label": "Sample interval (seconds)", "default": 5, "min": 1}, + "auto_restart": map[string]any{"type": "boolean", "label": "Auto-restart unhealthy containers", "default": true}, + "restart_cooldown_seconds": map[string]any{"type": "number", "label": "Restart cooldown (seconds)", "default": 120, "min": 10}, +} diff --git a/internal/observ/health.go b/internal/observ/health.go new file mode 100644 index 0000000..ab746cc --- /dev/null +++ b/internal/observ/health.go @@ -0,0 +1,241 @@ +package observ + +import ( + "context" + "encoding/json" + "os/exec" + "strings" + "sync" + "time" +) + +// Health status values as reported by Docker's HEALTHCHECK. +const ( + HealthHealthy = "healthy" + HealthUnhealthy = "unhealthy" + HealthStarting = "starting" + HealthNone = "none" +) + +// ContainerHealth is a container's current health, tagged with its deployment. +type ContainerHealth struct { + Container string `json:"container"` + Deployment string `json:"deployment"` + Status string `json:"status"` +} + +// HealthSource returns the current health of running containers. Injectable for tests. +type HealthSource func() ([]ContainerHealth, error) + +// RestartFunc restarts a container by name. Injectable for tests. +type RestartFunc func(container string) error + +// RecoveryEvent records a self-heal action for the UI/audit. +type RecoveryEvent struct { + Container string `json:"container"` + Deployment string `json:"deployment"` + At time.Time `json:"at"` +} + +// maxRestartAttempts bounds how many times a container is auto-restarted while it stays +// unhealthy, so a container a restart cannot fix is not restarted forever. The counter resets +// once the container reports healthy again. +const maxRestartAttempts = 3 + +// maxRecoveryEvents bounds the retained recovery history so a container that flaps for the +// life of the process cannot grow the slice without limit; only the most recent are kept. +const maxRecoveryEvents = 500 + +// HealthWatcher restarts running-but-unhealthy containers. It only acts on running containers +// (so it never revives a deployment the user intentionally stopped), only on deployments +// FlatRun manages, and only up to a bounded number of attempts per unhealthy streak. +type HealthWatcher struct { + source HealthSource + restart RestartFunc + interval time.Duration + cooldown time.Duration + now func() time.Time + + mu sync.Mutex + enabled bool + managed func(deployment string) bool + onRecover func(RecoveryEvent) + health map[string]ContainerHealth + lastHeal map[string]time.Time + attempts map[string]int + events []RecoveryEvent +} + +// SetManaged restricts auto-restart to deployments for which the predicate returns true. +// Health is still observed for all containers; only the restart action is scoped. +func (w *HealthWatcher) SetManaged(fn func(deployment string) bool) { + w.mu.Lock() + defer w.mu.Unlock() + w.managed = fn +} + +// OnRecover registers a callback fired after each auto-restart, so the core notification +// service can be told a container was recovered. +func (w *HealthWatcher) OnRecover(fn func(RecoveryEvent)) { + w.mu.Lock() + defer w.mu.Unlock() + w.onRecover = fn +} + +// SetEnabled turns auto-restart on or off. Health is still observed either way. +func (w *HealthWatcher) SetEnabled(on bool) { + w.mu.Lock() + defer w.mu.Unlock() + w.enabled = on +} + +func NewHealthWatcher(source HealthSource, restart RestartFunc, interval, cooldown time.Duration) *HealthWatcher { + if interval <= 0 { + interval = 15 * time.Second + } + if cooldown <= 0 { + cooldown = 2 * time.Minute + } + return &HealthWatcher{ + source: source, + restart: restart, + interval: interval, + cooldown: cooldown, + now: time.Now, + enabled: true, + health: map[string]ContainerHealth{}, + lastHeal: map[string]time.Time{}, + attempts: map[string]int{}, + } +} + +// checkOnce reads health once and restarts any unhealthy container past its cooldown. +func (w *HealthWatcher) checkOnce() { + states, err := w.source() + if err != nil { + return + } + w.mu.Lock() + defer w.mu.Unlock() + + w.health = make(map[string]ContainerHealth, len(states)) + for _, s := range states { + w.health[s.Container] = s + if s.Status == HealthHealthy { + delete(w.attempts, s.Container) + delete(w.lastHeal, s.Container) + } + if !w.enabled || s.Status != HealthUnhealthy { + continue + } + // Only act on deployments FlatRun manages, never arbitrary host containers. + if w.managed != nil && !w.managed(s.Deployment) { + continue + } + // Stop once repeated restarts have not fixed it; the container stays flagged + // unhealthy for the operator instead of being restarted forever. + if w.attempts[s.Container] >= maxRestartAttempts { + continue + } + last, seen := w.lastHeal[s.Container] + if seen && w.now().Sub(last) < w.cooldown { + continue + } + if err := w.restart(s.Container); err != nil { + continue + } + w.lastHeal[s.Container] = w.now() + w.attempts[s.Container]++ + ev := RecoveryEvent{Container: s.Container, Deployment: s.Deployment, At: w.now()} + w.events = append(w.events, ev) + if len(w.events) > maxRecoveryEvents { + w.events = w.events[len(w.events)-maxRecoveryEvents:] + } + if w.onRecover != nil { + go w.onRecover(ev) + } + } +} + +// Run checks health on each tick until ctx is cancelled. +func (w *HealthWatcher) Run(ctx context.Context) { + ticker := time.NewTicker(w.interval) + defer ticker.Stop() + w.checkOnce() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + w.checkOnce() + } + } +} + +// Snapshot returns the last-seen health per container. +func (w *HealthWatcher) Snapshot() []ContainerHealth { + w.mu.Lock() + defer w.mu.Unlock() + out := make([]ContainerHealth, 0, len(w.health)) + for _, h := range w.health { + out = append(out, h) + } + return out +} + +// Events returns the recovery actions taken, most recent last. Always non-nil so it +// serializes to a JSON array rather than null. +func (w *HealthWatcher) Events() []RecoveryEvent { + w.mu.Lock() + defer w.mu.Unlock() + out := make([]RecoveryEvent, 0, len(w.events)) + return append(out, w.events...) +} + +// DockerHealthSource reads container health via `docker ps`. +func DockerHealthSource() ([]ContainerHealth, error) { + deployments := containerDeployments() + out, err := exec.Command("docker", "ps", "--format", "{{json .}}").Output() + if err != nil { + return nil, err + } + var states []ContainerHealth + for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") { + if line == "" { + continue + } + var raw struct { + Names string `json:"Names"` + Status string `json:"Status"` + } + if err := json.Unmarshal([]byte(line), &raw); err != nil { + continue + } + states = append(states, ContainerHealth{ + Container: raw.Names, + Deployment: deployments[raw.Names], + Status: healthFromStatus(raw.Status), + }) + } + return states, nil +} + +// healthFromStatus reads the "(healthy)" / "(unhealthy)" / "(health: starting)" suffix +// Docker appends to a running container's Status string. +func healthFromStatus(status string) string { + switch { + case strings.Contains(status, "(unhealthy)"): + return HealthUnhealthy + case strings.Contains(status, "(healthy)"): + return HealthHealthy + case strings.Contains(status, "health: starting"): + return HealthStarting + default: + return HealthNone + } +} + +// DockerRestart restarts a container by name. +func DockerRestart(container string) error { + return exec.Command("docker", "restart", container).Run() +} diff --git a/internal/observ/health_test.go b/internal/observ/health_test.go new file mode 100644 index 0000000..68246e5 --- /dev/null +++ b/internal/observ/health_test.go @@ -0,0 +1,173 @@ +package observ + +import ( + "testing" + "time" +) + +func TestHealthWatcherRestartsUnhealthy(t *testing.T) { + states := []ContainerHealth{ + {Container: "shop-web", Deployment: "shop", Status: HealthUnhealthy}, + {Container: "shop-db", Deployment: "shop", Status: HealthHealthy}, + } + var restarted []string + w := NewHealthWatcher( + func() ([]ContainerHealth, error) { return states, nil }, + func(c string) error { restarted = append(restarted, c); return nil }, + time.Second, time.Minute, + ) + w.now = func() time.Time { return time.Unix(1_700_000_000, 0) } + + w.checkOnce() + + if len(restarted) != 1 || restarted[0] != "shop-web" { + t.Fatalf("expected only shop-web restarted, got %v", restarted) + } + if evs := w.Events(); len(evs) != 1 || evs[0].Container != "shop-web" { + t.Errorf("expected one recovery event for shop-web, got %+v", evs) + } +} + +func TestHealthWatcherHonorsCooldown(t *testing.T) { + states := []ContainerHealth{{Container: "web", Deployment: "d", Status: HealthUnhealthy}} + count := 0 + now := time.Unix(1_700_000_000, 0) + w := NewHealthWatcher( + func() ([]ContainerHealth, error) { return states, nil }, + func(string) error { count++; return nil }, + time.Second, 2*time.Minute, + ) + w.now = func() time.Time { return now } + + w.checkOnce() // restarts + w.checkOnce() // within cooldown, must not restart again + if count != 1 { + t.Fatalf("expected 1 restart within cooldown, got %d", count) + } + + now = now.Add(3 * time.Minute) // past cooldown + w.checkOnce() + if count != 2 { + t.Fatalf("expected a second restart after cooldown, got %d", count) + } +} + +func TestHealthWatcherNeverActsOnStoppedContainers(t *testing.T) { + // A user-stopped container is not running, so it never appears in the source; the + // watcher therefore cannot restart it. + running := []ContainerHealth{{Container: "web", Deployment: "d", Status: HealthHealthy}} + restarts := 0 + w := NewHealthWatcher( + func() ([]ContainerHealth, error) { return running, nil }, + func(string) error { restarts++; return nil }, + time.Second, time.Minute, + ) + w.checkOnce() + if restarts != 0 { + t.Errorf("no running container is unhealthy, expected 0 restarts, got %d", restarts) + } +} + +func TestHealthWatcherOnlyRestartsManaged(t *testing.T) { + states := []ContainerHealth{ + {Container: "app-web", Deployment: "myapp", Status: HealthUnhealthy}, + {Container: "pagemind-cli-1", Deployment: "pagemind", Status: HealthUnhealthy}, + } + var restarted []string + w := NewHealthWatcher( + func() ([]ContainerHealth, error) { return states, nil }, + func(c string) error { restarted = append(restarted, c); return nil }, + time.Second, time.Minute, + ) + w.SetManaged(func(dep string) bool { return dep == "myapp" }) + + w.checkOnce() + if len(restarted) != 1 || restarted[0] != "app-web" { + t.Fatalf("only the managed deployment should be restarted, got %v", restarted) + } +} + +func TestHealthWatcherGivesUpAfterMaxAttempts(t *testing.T) { + states := []ContainerHealth{{Container: "web", Deployment: "d", Status: HealthUnhealthy}} + count := 0 + now := time.Unix(1_700_000_000, 0) + w := NewHealthWatcher( + func() ([]ContainerHealth, error) { return states, nil }, + func(string) error { count++; return nil }, + time.Second, time.Minute, + ) + w.now = func() time.Time { return now } + + // Each check is past the cooldown; it should restart only up to the cap. + for i := 0; i < 6; i++ { + w.checkOnce() + now = now.Add(2 * time.Minute) + } + if count != maxRestartAttempts { + t.Fatalf("expected at most %d restarts, got %d", maxRestartAttempts, count) + } + + // Once healthy, the counter resets and it will act again if it fails later. + states[0].Status = HealthHealthy + w.checkOnce() + states[0].Status = HealthUnhealthy + now = now.Add(2 * time.Minute) + w.checkOnce() + if count != maxRestartAttempts+1 { + t.Errorf("counter should reset after a healthy report, got %d restarts", count) + } +} + +func TestHealthWatcherFiresOnRecover(t *testing.T) { + states := []ContainerHealth{{Container: "web", Deployment: "shop", Status: HealthUnhealthy}} + w := NewHealthWatcher( + func() ([]ContainerHealth, error) { return states, nil }, + func(string) error { return nil }, + time.Second, time.Minute, + ) + got := make(chan RecoveryEvent, 1) + w.OnRecover(func(ev RecoveryEvent) { got <- ev }) + + w.checkOnce() + select { + case ev := <-got: + if ev.Container != "web" || ev.Deployment != "shop" { + t.Errorf("unexpected recover event %+v", ev) + } + case <-time.After(time.Second): + t.Fatal("OnRecover was not fired on restart") + } +} + +func TestHealthWatcherDisabledObservesButDoesNotRestart(t *testing.T) { + states := []ContainerHealth{{Container: "web", Deployment: "d", Status: HealthUnhealthy}} + restarts := 0 + w := NewHealthWatcher( + func() ([]ContainerHealth, error) { return states, nil }, + func(string) error { restarts++; return nil }, + time.Second, time.Minute, + ) + w.SetEnabled(false) + w.checkOnce() + + if restarts != 0 { + t.Errorf("disabled watcher must not restart, got %d", restarts) + } + if snap := w.Snapshot(); len(snap) != 1 || snap[0].Status != HealthUnhealthy { + t.Errorf("disabled watcher should still observe health, got %+v", snap) + } +} + +func TestHealthFromStatus(t *testing.T) { + cases := map[string]string{ + "Up 2 hours (healthy)": HealthHealthy, + "Up 5 minutes (unhealthy)": HealthUnhealthy, + "Up 3 seconds (health: starting)": HealthStarting, + "Up 10 days": HealthNone, + } + for in, want := range cases { + if got := healthFromStatus(in); got != want { + t.Errorf("healthFromStatus(%q) = %q, want %q", in, got, want) + } + } +} diff --git a/internal/observ/plugin.go b/internal/observ/plugin.go new file mode 100644 index 0000000..217d024 --- /dev/null +++ b/internal/observ/plugin.go @@ -0,0 +1,90 @@ +package observ + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "path/filepath" + "time" + + "github.com/flatrun/agent/pkg/pluginapi" + "github.com/flatrun/agent/pkg/pluginsdk" +) + +// PluginInfo identifies the observability app to the host and declares the UI it contributes: +// a metrics + health panel inside each deployment, and a settings form for its config. +var PluginInfo = pluginapi.Info{ + Name: "observability", + Version: "0.1.0", + DisplayName: "Observability", + Description: "Per-deployment metrics and health, OpenTelemetry-native.", + Capabilities: []string{"metrics", "docker"}, + ConfigSchema: ConfigSchema, + UIExtensions: []pluginapi.UIExtension{ + {Slot: "deployment.detail", Kind: "metrics-panel", Title: "Metrics & Health", Icon: "activity", Endpoint: "/metrics/deployment"}, + {Slot: "settings", Kind: "form", Title: "Observability", Icon: "activity", Endpoint: "/config"}, + }, +} + +// RunPlugin collects metrics, watches container health, restarts unhealthy containers, and +// serves it all until the host stops the process. It is the entry point for both the +// standalone plugin binary and the agent's self-exec subcommand. +func RunPlugin() error { + ctx := context.Background() + cfgStore := NewConfigStore(os.Getenv(pluginapi.EnvDataDir)) + cfg := cfgStore.Load() + + dataDir := os.Getenv(pluginapi.EnvDataDir) + store := NewStore(720) + collector := NewCollector(store, DockerStatsSource, cfg.sampleInterval()) + watcher := NewHealthWatcher(DockerHealthSource, DockerRestart, 15*time.Second, cfg.restartCooldown()) + watcher.SetEnabled(cfg.AutoRestart) + // A deployment FlatRun manages has a directory under the deployments path; only those are + // eligible for auto-restart, so stray host containers are never touched. + watcher.SetManaged(func(deployment string) bool { + if deployment == "" || dataDir == "" { + return false + } + info, err := os.Stat(filepath.Join(dataDir, deployment)) + return err == nil && info.IsDir() + }) + + watcher.OnRecover(func(ev RecoveryEvent) { + emitNotification( + fmt.Sprintf("Auto-recovered %s", ev.Container), + fmt.Sprintf("Container %s in deployment %s was unhealthy and has been restarted.", ev.Container, ev.Deployment), + ) + }) + + go collector.Run(ctx) + go watcher.Run(ctx) + + applyConfig := func(c Config) { + watcher.SetEnabled(c.AutoRestart) + } + + return pluginsdk.Serve(PluginInfo, Handler(store, watcher, cfgStore, applyConfig), buildTools(store, watcher, cfgStore)...) +} + +// emitNotification asks the core to deliver a notification to the operator's configured +// targets. Delivery config and routing live in the agent, not the plugin. +func emitNotification(title, message string) { + base, token := pluginsdk.AgentCallback() + if base == "" || token == "" { + return + } + body, _ := json.Marshal(map[string]string{"title": title, "message": message}) + req, err := http.NewRequest(http.MethodPost, base+"/internal/notify/emit", bytes.NewReader(body)) + if err != nil { + return + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Plugin-Token", token) + client := &http.Client{Timeout: 5 * time.Second} + if resp, err := client.Do(req); err == nil { + _ = resp.Body.Close() + } +} diff --git a/internal/observ/store.go b/internal/observ/store.go new file mode 100644 index 0000000..0cfd656 --- /dev/null +++ b/internal/observ/store.go @@ -0,0 +1,220 @@ +// Package observ is the FlatRun observability engine: it collects per-container metrics, +// keeps a bounded recent-history window for the native UI, and (elsewhere) exposes them over +// OTLP so any OpenTelemetry-compatible tool can consume the same data. Metric names follow +// the OpenTelemetry container semantic conventions. +package observ + +import ( + "sort" + "sync" + "time" +) + +// OpenTelemetry container metric names (semconv). Emitting these verbatim keeps FlatRun's +// metrics interoperable with any OTel backend. +const ( + MetricCPUUsage = "container.cpu.usage" + MetricMemoryUsage = "container.memory.usage" + MetricMemoryLimit = "container.memory.limit" + MetricNetworkRx = "container.network.io.rx" + MetricNetworkTx = "container.network.io.tx" +) + +// Sample is a single metric reading at a point in time. +type Sample struct { + Time time.Time `json:"time"` + Value float64 `json:"value"` +} + +// SeriesKey identifies one metric stream: a metric for a container within a deployment. +type SeriesKey struct { + Deployment string + Container string + Metric string +} + +// ContainerSample is the raw per-container reading the scraper feeds in; the store expands +// it into the individual semconv metric series. +type ContainerSample struct { + Deployment string + Container string + CPUPercent float64 + MemoryUsage uint64 + MemoryLimit uint64 + NetworkRx uint64 + NetworkTx uint64 +} + +// Store holds a bounded ring of recent samples per series, so the UI can render recent +// history without an external time-series database. Older samples are discarded once the +// per-series capacity is reached. +type Store struct { + mu sync.RWMutex + capacity int + retention time.Duration + series map[SeriesKey]*ring + lastSweep time.Time +} + +func NewStore(capacityPerSeries int) *Store { + if capacityPerSeries <= 0 { + capacityPerSeries = 720 // e.g. 1h at 5s resolution + } + return &Store{ + capacity: capacityPerSeries, + retention: time.Hour, + series: make(map[SeriesKey]*ring), + } +} + +// sweepStale drops series whose newest sample has aged past the retention window, so a +// container that stops reporting (removed, deployment torn down) does not keep its ring alive +// for the process lifetime. The caller must hold s.mu. It runs at most once per retention +// window using the sample clock, keeping it O(series) amortized. +func (s *Store) sweepStale(now time.Time) { + if s.retention <= 0 { + return + } + if !s.lastSweep.IsZero() && now.Sub(s.lastSweep) < s.retention { + return + } + s.lastSweep = now + cutoff := now.Add(-s.retention) + for k, r := range s.series { + if last, ok := r.last(); ok && last.Time.Before(cutoff) { + delete(s.series, k) + } + } +} + +func (s *Store) add(key SeriesKey, sample Sample) { + r := s.series[key] + if r == nil { + r = newRing(s.capacity) + s.series[key] = r + } + r.push(sample) +} + +// Record expands a container reading into its semconv series and stores each at t. +func (s *Store) Record(c ContainerSample, t time.Time) { + s.mu.Lock() + defer s.mu.Unlock() + base := SeriesKey{Deployment: c.Deployment, Container: c.Container} + for metric, value := range map[string]float64{ + MetricCPUUsage: c.CPUPercent, + MetricMemoryUsage: float64(c.MemoryUsage), + MetricMemoryLimit: float64(c.MemoryLimit), + MetricNetworkRx: float64(c.NetworkRx), + MetricNetworkTx: float64(c.NetworkTx), + } { + key := base + key.Metric = metric + s.add(key, Sample{Time: t, Value: value}) + } + s.sweepStale(t) +} + +// Range returns the samples for a series recorded at or after since, oldest first. +func (s *Store) Range(key SeriesKey, since time.Time) []Sample { + s.mu.RLock() + defer s.mu.RUnlock() + r := s.series[key] + if r == nil { + return nil + } + return r.since(since) +} + +// LatestPoint is a series' most recent sample. +type LatestPoint struct { + SeriesKey + Sample +} + +// Latest returns the most recent sample of every series, sorted by key. +func (s *Store) Latest() []LatestPoint { + s.mu.RLock() + defer s.mu.RUnlock() + out := make([]LatestPoint, 0, len(s.series)) + for k, r := range s.series { + if last, ok := r.last(); ok { + out = append(out, LatestPoint{SeriesKey: k, Sample: last}) + } + } + sort.Slice(out, func(i, j int) bool { + if out[i].Deployment != out[j].Deployment { + return out[i].Deployment < out[j].Deployment + } + if out[i].Container != out[j].Container { + return out[i].Container < out[j].Container + } + return out[i].Metric < out[j].Metric + }) + return out +} + +// Series lists the keys currently held, so callers can enumerate what is available. +func (s *Store) Series() []SeriesKey { + s.mu.RLock() + defer s.mu.RUnlock() + keys := make([]SeriesKey, 0, len(s.series)) + for k := range s.series { + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { + if keys[i].Deployment != keys[j].Deployment { + return keys[i].Deployment < keys[j].Deployment + } + if keys[i].Container != keys[j].Container { + return keys[i].Container < keys[j].Container + } + return keys[i].Metric < keys[j].Metric + }) + return keys +} + +// ring is a fixed-capacity circular buffer of samples in insertion order. +type ring struct { + buf []Sample + next int + count int +} + +func newRing(capacity int) *ring { + return &ring{buf: make([]Sample, capacity)} +} + +func (r *ring) push(s Sample) { + r.buf[r.next] = s + r.next = (r.next + 1) % len(r.buf) + if r.count < len(r.buf) { + r.count++ + } +} + +// ordered returns the buffered samples oldest-first. +func (r *ring) ordered() []Sample { + out := make([]Sample, 0, r.count) + start := 0 + if r.count == len(r.buf) { + start = r.next + } + for i := 0; i < r.count; i++ { + out = append(out, r.buf[(start+i)%len(r.buf)]) + } + return out +} + +func (r *ring) last() (Sample, bool) { + if r.count == 0 { + return Sample{}, false + } + return r.buf[(r.next-1+len(r.buf))%len(r.buf)], true +} + +func (r *ring) since(t time.Time) []Sample { + all := r.ordered() + idx := sort.Search(len(all), func(i int) bool { return !all[i].Time.Before(t) }) + return all[idx:] +} diff --git a/internal/observ/store_test.go b/internal/observ/store_test.go new file mode 100644 index 0000000..3ae0839 --- /dev/null +++ b/internal/observ/store_test.go @@ -0,0 +1,92 @@ +package observ + +import ( + "testing" + "time" +) + +func TestStoreRecordExpandsSemconvSeries(t *testing.T) { + s := NewStore(10) + t0 := time.Unix(1_700_000_000, 0) + s.Record(ContainerSample{ + Deployment: "shop", Container: "shop-web", + CPUPercent: 12.5, MemoryUsage: 100, MemoryLimit: 1000, NetworkRx: 5, NetworkTx: 7, + }, t0) + + keys := s.Series() + if len(keys) != 5 { + t.Fatalf("expected 5 semconv series, got %d: %+v", len(keys), keys) + } + + cpu := s.Range(SeriesKey{Deployment: "shop", Container: "shop-web", Metric: MetricCPUUsage}, t0) + if len(cpu) != 1 || cpu[0].Value != 12.5 { + t.Errorf("cpu series = %+v, want one sample of 12.5", cpu) + } + mem := s.Range(SeriesKey{Deployment: "shop", Container: "shop-web", Metric: MetricMemoryUsage}, t0) + if len(mem) != 1 || mem[0].Value != 100 { + t.Errorf("memory series = %+v, want one sample of 100", mem) + } +} + +func TestStoreRingEvictsOldest(t *testing.T) { + s := NewStore(3) + key := SeriesKey{Deployment: "d", Container: "c", Metric: MetricCPUUsage} + base := time.Unix(1_700_000_000, 0) + for i := 0; i < 5; i++ { + s.Record(ContainerSample{Deployment: "d", Container: "c", CPUPercent: float64(i)}, base.Add(time.Duration(i)*time.Second)) + } + + got := s.Range(key, base) + if len(got) != 3 { + t.Fatalf("expected ring to hold 3, got %d", len(got)) + } + // Oldest two (0,1) evicted; remaining are 2,3,4 in order. + for i, want := range []float64{2, 3, 4} { + if got[i].Value != want { + t.Errorf("sample[%d] = %v, want %v", i, got[i].Value, want) + } + } +} + +func TestStoreRangeSinceFilters(t *testing.T) { + s := NewStore(10) + key := SeriesKey{Deployment: "d", Container: "c", Metric: MetricCPUUsage} + base := time.Unix(1_700_000_000, 0) + for i := 0; i < 5; i++ { + s.Record(ContainerSample{Deployment: "d", Container: "c", CPUPercent: float64(i)}, base.Add(time.Duration(i)*time.Second)) + } + + got := s.Range(key, base.Add(3*time.Second)) + if len(got) != 2 || got[0].Value != 3 || got[1].Value != 4 { + t.Errorf("since(t+3s) = %+v, want samples 3 and 4", got) + } +} + +func TestStoreEvictsStaleSeries(t *testing.T) { + s := NewStore(10) + base := time.Unix(1_700_000_000, 0) + s.Record(ContainerSample{Deployment: "gone", Container: "gone-web", CPUPercent: 1}, base) + if len(s.Series()) != 5 { + t.Fatalf("expected 5 series after first record, got %d", len(s.Series())) + } + + // A live container reports well past the retention window; the departed one must be dropped. + s.Record(ContainerSample{Deployment: "live", Container: "live-web", CPUPercent: 2}, base.Add(2*time.Hour)) + + for _, k := range s.Series() { + if k.Deployment == "gone" { + t.Errorf("stale series was not evicted: %+v", k) + } + } + if len(s.Series()) != 5 { + t.Errorf("expected only the live deployment's 5 series to remain, got %d", len(s.Series())) + } +} + +func TestStoreRangeUnknownSeriesIsEmpty(t *testing.T) { + s := NewStore(10) + got := s.Range(SeriesKey{Deployment: "x", Container: "y", Metric: MetricCPUUsage}, time.Unix(0, 0)) + if got != nil { + t.Errorf("unknown series should return nil, got %+v", got) + } +} diff --git a/internal/observ/timeseries.go b/internal/observ/timeseries.go new file mode 100644 index 0000000..b4aea95 --- /dev/null +++ b/internal/observ/timeseries.go @@ -0,0 +1,76 @@ +package observ + +import ( + "sort" + "time" +) + +// MetricSeries is one metric's aligned time series across containers: a shared timestamp +// axis and, per container, a value at each timestamp (nil for gaps). This is the shape a +// time-series chart consumes directly. +type MetricSeries struct { + Containers []string `json:"containers"` + Timestamps []int64 `json:"timestamps"` // unix seconds, ascending + Values [][]*float64 `json:"values"` // [container][timestamp] +} + +// TimeSeriesResponse is the batch of a deployment's metric series over a window. +type TimeSeriesResponse struct { + Deployment string `json:"deployment"` + Metrics map[string]MetricSeries `json:"metrics"` +} + +// buildTimeSeries gathers every series for a deployment (all if empty) since a time and +// aligns each metric's containers onto a shared timestamp axis. +func buildTimeSeries(store *Store, deployment string, since time.Time) map[string]MetricSeries { + grouped := map[string]map[string][]Sample{} + for _, key := range store.Series() { + if deployment != "" && key.Deployment != deployment { + continue + } + samples := store.Range(key, since) + if len(samples) == 0 { + continue + } + if grouped[key.Metric] == nil { + grouped[key.Metric] = map[string][]Sample{} + } + grouped[key.Metric][key.Container] = samples + } + + out := make(map[string]MetricSeries, len(grouped)) + for metric, byContainer := range grouped { + containers := make([]string, 0, len(byContainer)) + tsSet := map[int64]struct{}{} + for c, samples := range byContainer { + containers = append(containers, c) + for _, s := range samples { + tsSet[s.Time.Unix()] = struct{}{} + } + } + sort.Strings(containers) + timestamps := make([]int64, 0, len(tsSet)) + for t := range tsSet { + timestamps = append(timestamps, t) + } + sort.Slice(timestamps, func(i, j int) bool { return timestamps[i] < timestamps[j] }) + + values := make([][]*float64, len(containers)) + for ci, c := range containers { + atTime := make(map[int64]float64, len(byContainer[c])) + for _, s := range byContainer[c] { + atTime[s.Time.Unix()] = s.Value + } + row := make([]*float64, len(timestamps)) + for ti, t := range timestamps { + if v, ok := atTime[t]; ok { + vv := v + row[ti] = &vv + } + } + values[ci] = row + } + out[metric] = MetricSeries{Containers: containers, Timestamps: timestamps, Values: values} + } + return out +} diff --git a/internal/observ/timeseries_test.go b/internal/observ/timeseries_test.go new file mode 100644 index 0000000..d8943bd --- /dev/null +++ b/internal/observ/timeseries_test.go @@ -0,0 +1,61 @@ +package observ + +import ( + "testing" + "time" +) + +func TestBuildTimeSeriesAlignsContainers(t *testing.T) { + store := NewStore(100) + base := time.Unix(1_700_000_000, 0) + // Two containers sampled at the same three ticks. + for i := 0; i < 3; i++ { + at := base.Add(time.Duration(i) * time.Second) + store.Record(ContainerSample{Deployment: "shop", Container: "shop-web", CPUPercent: float64(i)}, at) + store.Record(ContainerSample{Deployment: "shop", Container: "shop-db", CPUPercent: float64(i * 2)}, at) + } + + ms := buildTimeSeries(store, "shop", base.Add(-time.Minute)) + cpu, ok := ms[MetricCPUUsage] + if !ok { + t.Fatalf("cpu series missing; got metrics %v", keys(ms)) + } + if len(cpu.Containers) != 2 { + t.Fatalf("expected 2 containers, got %v", cpu.Containers) + } + if len(cpu.Timestamps) != 3 { + t.Fatalf("expected 3 aligned timestamps, got %d", len(cpu.Timestamps)) + } + // Values are [container][timestamp], aligned to the shared axis. + web := cpu.Containers[0] // "shop-db" sorts first + _ = web + for ci, name := range cpu.Containers { + if len(cpu.Values[ci]) != len(cpu.Timestamps) { + t.Errorf("container %s row length %d != timestamps %d", name, len(cpu.Values[ci]), len(cpu.Timestamps)) + } + } +} + +func TestBuildTimeSeriesFiltersDeployment(t *testing.T) { + store := NewStore(100) + now := time.Unix(1_700_000_000, 0) + store.Record(ContainerSample{Deployment: "shop", Container: "shop-web", CPUPercent: 1}, now) + store.Record(ContainerSample{Deployment: "blog", Container: "blog-web", CPUPercent: 2}, now) + + ms := buildTimeSeries(store, "shop", now.Add(-time.Minute)) + for _, s := range ms { + for _, c := range s.Containers { + if c == "blog-web" { + t.Errorf("blog container leaked into shop series") + } + } + } +} + +func keys(m map[string]MetricSeries) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return out +} diff --git a/internal/observ/tools.go b/internal/observ/tools.go new file mode 100644 index 0000000..aad94cf --- /dev/null +++ b/internal/observ/tools.go @@ -0,0 +1,173 @@ +package observ + +import ( + "fmt" + "os/exec" + "strings" + + "github.com/flatrun/agent/pkg/pluginapi" + "github.com/flatrun/agent/pkg/pluginsdk" +) + +// containerProject resolves the compose project (deployment) a container belongs to. It is a +// package var so tests can stub it without Docker. +var containerProject = dockerContainerProject + +func dockerContainerProject(container string) (string, error) { + out, err := exec.Command("docker", "inspect", "-f", + `{{ index .Config.Labels "com.docker.compose.project" }}`, container).Output() + if err != nil { + return "", err + } + return strings.TrimSpace(string(out)), nil +} + +// buildTools exposes observability capabilities to the AI assistant: read tools for health +// and metrics, and mutating tools to manage recovery. Mutating tools set Mutates so the host +// gates them on write access. +func buildTools(store *Store, watcher *HealthWatcher, cfgStore *ConfigStore) []pluginsdk.Tool { + strParam := func(name, desc string, required bool) map[string]any { + schema := map[string]any{ + "type": "object", + "properties": map[string]any{name: map[string]any{"type": "string", "description": desc}}, + } + if required { + schema["required"] = []string{name} + } + return schema + } + + return []pluginsdk.Tool{ + { + Spec: pluginapi.ToolSpec{ + Name: "get_deployment_health", + Description: "Report the health (healthy/unhealthy/starting) of each container in a deployment.", + Parameters: strParam("deployment", "The deployment name.", true), + }, + Run: func(args map[string]any) (string, error) { + dep := argStr(args, "deployment") + var b strings.Builder + found := false + for _, h := range watcher.Snapshot() { + if dep != "" && h.Deployment != dep { + continue + } + found = true + fmt.Fprintf(&b, "%s: %s\n", h.Container, h.Status) + } + if !found { + return fmt.Sprintf("No running containers found for deployment %q.", dep), nil + } + return b.String(), nil + }, + }, + { + Spec: pluginapi.ToolSpec{ + Name: "get_deployment_metrics", + Description: "Summarize the latest CPU, memory and network usage per container in a deployment.", + Parameters: strParam("deployment", "The deployment name.", true), + }, + Run: func(args map[string]any) (string, error) { + dep := argStr(args, "deployment") + groups := filterByDeployment(groupLatest(store.Latest()), dep) + if len(groups) == 0 { + return fmt.Sprintf("No metrics for deployment %q yet.", dep), nil + } + var b strings.Builder + for _, g := range groups { + for _, c := range g.Containers { + fmt.Fprintf(&b, "%s: cpu %.1f%%, mem %s, net rx %s / tx %s\n", + c.Container, + c.Metrics[MetricCPUUsage], + humanBytes(c.Metrics[MetricMemoryUsage]), + humanBytes(c.Metrics[MetricNetworkRx]), + humanBytes(c.Metrics[MetricNetworkTx]), + ) + } + } + return b.String(), nil + }, + }, + { + Spec: pluginapi.ToolSpec{ + Name: "set_auto_restart", + Description: "Enable or disable automatically restarting unhealthy containers.", + Parameters: map[string]any{ + "type": "object", + "properties": map[string]any{"enabled": map[string]any{"type": "boolean", "description": "Whether auto-restart is on."}}, + "required": []string{"enabled"}, + }, + Mutates: true, + }, + Run: func(args map[string]any) (string, error) { + enabled, _ := args["enabled"].(bool) + cfg := cfgStore.Load() + cfg.AutoRestart = enabled + if err := cfgStore.Save(cfg); err != nil { + return "", err + } + watcher.SetEnabled(enabled) + return fmt.Sprintf("Auto-restart is now %s.", onOff(enabled)), nil + }, + }, + { + Spec: pluginapi.ToolSpec{ + Name: "restart_container", + Description: "Restart a container by name.", + Parameters: strParam("container", "The container name to restart.", true), + Mutates: true, + }, + Run: func(args map[string]any) (string, error) { + container := argStr(args, "container") + if container == "" { + return "", fmt.Errorf("no container specified") + } + // _deployment is set by the agent to the write-authorized deployment; the + // container must belong to it, so a caller cannot restart another + // deployment's container by naming it. + scope := argStr(args, "_deployment") + if scope == "" { + return "", fmt.Errorf("restarting a container requires a deployment-scoped session") + } + project, err := containerProject(container) + if err != nil { + return "", fmt.Errorf("could not resolve container %q: %w", container, err) + } + if project != scope { + return "", fmt.Errorf("container %q is not part of deployment %q", container, scope) + } + if err := DockerRestart(container); err != nil { + return "", fmt.Errorf("restart failed: %w", err) + } + return fmt.Sprintf("Restarted %s.", container), nil + }, + }, + } +} + +func argStr(args map[string]any, key string) string { + if v, ok := args[key].(string); ok { + return v + } + return "" +} + +func onOff(b bool) string { + if b { + return "on" + } + return "off" +} + +func humanBytes(v float64) string { + switch { + case v >= 1<<30: + return fmt.Sprintf("%.1fG", v/(1<<30)) + case v >= 1<<20: + return fmt.Sprintf("%.0fM", v/(1<<20)) + case v >= 1<<10: + return fmt.Sprintf("%.0fK", v/(1<<10)) + default: + return fmt.Sprintf("%.0fB", v) + } +} diff --git a/internal/observ/tools_test.go b/internal/observ/tools_test.go new file mode 100644 index 0000000..58a834d --- /dev/null +++ b/internal/observ/tools_test.go @@ -0,0 +1,64 @@ +package observ + +import ( + "strings" + "testing" + "time" + + "github.com/flatrun/agent/pkg/pluginsdk" +) + +func findTool(tools []pluginsdk.Tool, name string) (pluginsdk.Tool, bool) { + for _, t := range tools { + if t.Spec.Name == name { + return t, true + } + } + return pluginsdk.Tool{}, false +} + +func newTools(t *testing.T) []pluginsdk.Tool { + store := NewStore(10) + watcher := NewHealthWatcher(func() ([]ContainerHealth, error) { return nil, nil }, func(string) error { return nil }, time.Second, time.Minute) + return buildTools(store, watcher, NewConfigStore(t.TempDir())) +} + +func TestRestartContainerRequiresScope(t *testing.T) { + restart, ok := findTool(newTools(t), "restart_container") + if !ok { + t.Fatal("restart_container tool missing") + } + if !restart.Spec.Mutates { + t.Error("restart_container must be marked as mutating") + } + // No scope (unscoped session) must be refused before any docker action. + if _, err := restart.Run(map[string]any{"container": "x"}); err == nil || !strings.Contains(err.Error(), "deployment-scoped") { + t.Errorf("expected scope-required error, got %v", err) + } +} + +func TestRestartContainerRefusesCrossDeployment(t *testing.T) { + orig := containerProject + defer func() { containerProject = orig }() + // The container actually belongs to another deployment. + containerProject = func(string) (string, error) { return "other-app", nil } + + restart, _ := findTool(newTools(t), "restart_container") + _, err := restart.Run(map[string]any{"container": "victim", "_deployment": "myapp"}) + if err == nil || !strings.Contains(err.Error(), "not part of deployment") { + t.Errorf("expected cross-deployment refusal, got %v", err) + } +} + +func TestHealthAndMetricsToolsAreReadOnly(t *testing.T) { + tools := newTools(t) + for _, name := range []string{"get_deployment_health", "get_deployment_metrics"} { + tl, ok := findTool(tools, name) + if !ok { + t.Fatalf("%s missing", name) + } + if tl.Spec.Mutates { + t.Errorf("%s should not be mutating", name) + } + } +} diff --git a/internal/pluginhost/host.go b/internal/pluginhost/host.go index 940776e..8ebebf9 100644 --- a/internal/pluginhost/host.go +++ b/internal/pluginhost/host.go @@ -3,6 +3,7 @@ package pluginhost import ( + "bytes" "context" "crypto/rand" "encoding/hex" @@ -29,36 +30,62 @@ type managed struct { proxy *httputil.ReverseProxy } +type builtin struct { + name string + path string + args []string +} + type Host struct { pluginsDir string runtimeDir string agentURL string token string + dataDir string handshake string + builtins []builtin mu sync.RWMutex running map[string]*managed } func New(pluginsDir, runtimeDir, agentURL, token string) *Host { + // pluginsDir is /.flatrun/plugins, so the deployments base is two levels up. + dataDir := "" + if pluginsDir != "" { + dataDir = filepath.Dir(filepath.Dir(pluginsDir)) + } return &Host{ pluginsDir: pluginsDir, runtimeDir: runtimeDir, agentURL: agentURL, token: token, + dataDir: dataDir, handshake: randomCookie(), running: make(map[string]*managed), } } -// Start launches every plugin binary; a plugin that fails to launch is logged and skipped. +// Builtin registers a plugin shipped inside the agent, launched by running the given command +// (typically the agent re-execing itself with a subcommand). Call before Start. +func (h *Host) Builtin(name, path string, args ...string) { + h.builtins = append(h.builtins, builtin{name: name, path: path, args: args}) +} + +// Start launches the built-in plugins and every external plugin binary; a plugin that fails +// to launch is logged and skipped. func (h *Host) Start() error { - if h.pluginsDir == "" { - return nil - } if err := os.MkdirAll(h.runtimeDir, 0755); err != nil { return err } + for _, b := range h.builtins { + if err := h.launch(b.name, exec.Command(b.path, b.args...)); err != nil { + log.Printf("[pluginhost] built-in %s failed to start: %v", b.name, err) + } + } + if h.pluginsDir == "" { + return nil + } entries, err := os.ReadDir(h.pluginsDir) if err != nil { if os.IsNotExist(err) { @@ -74,23 +101,23 @@ func (h *Host) Start() error { if !isExecutable(bin) { continue } - if err := h.launch(entry.Name(), bin); err != nil { + if err := h.launch(entry.Name(), exec.Command(bin)); err != nil { log.Printf("[pluginhost] %s failed to start: %v", entry.Name(), err) } } return nil } -func (h *Host) launch(name, bin string) error { +func (h *Host) launch(name string, cmd *exec.Cmd) error { socket := filepath.Join(h.runtimeDir, name+".sock") _ = os.Remove(socket) - cmd := exec.Command(bin) cmd.Env = append(os.Environ(), pluginapi.EnvSocket+"="+socket, pluginapi.EnvHandshake+"="+h.handshake, pluginapi.EnvAgentURL+"="+h.agentURL, pluginapi.EnvToken+"="+h.token, + pluginapi.EnvDataDir+"="+h.dataDir, ) cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr @@ -157,6 +184,36 @@ func (h *Host) Proxy(name string) (*httputil.ReverseProxy, bool) { return m.proxy, true } +// ExecTool invokes a tool on a running plugin over its socket and returns the textual result. +func (h *Host) ExecTool(plugin, tool string, args map[string]any) (string, error) { + h.mu.RLock() + m, ok := h.running[plugin] + h.mu.RUnlock() + if !ok { + return "", fmt.Errorf("plugin %q not running", plugin) + } + + body, _ := json.Marshal(map[string]any{"name": tool, "args": args}) + client := &http.Client{Transport: unixTransport(m.socket), Timeout: 30 * time.Second} + resp, err := client.Post("http://plugin"+pluginapi.ToolExecPath, "application/json", bytes.NewReader(body)) + if err != nil { + return "", err + } + defer resp.Body.Close() + + var out struct { + Result string `json:"result"` + Error string `json:"error"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return "", err + } + if out.Error != "" { + return "", fmt.Errorf("%s", out.Error) + } + return out.Result, nil +} + func (h *Host) Infos() []pluginapi.Info { h.mu.RLock() defer h.mu.RUnlock() diff --git a/internal/pluginhost/host_test.go b/internal/pluginhost/host_test.go index 690dee5..982dd34 100644 --- a/internal/pluginhost/host_test.go +++ b/internal/pluginhost/host_test.go @@ -88,3 +88,84 @@ func TestHostLaunchesAndProxiesBinaryPlugin(t *testing.T) { t.Error("plugin should be gone after Stop") } } + +// A plugin's declared tools are advertised in its Info and invokable via ExecTool. +func TestHostExecToolInvokesPluginTool(t *testing.T) { + if testing.Short() { + t.Skip("skipping subprocess build in short mode") + } + goBin, err := exec.LookPath("go") + if err != nil { + t.Skip("go toolchain not available") + } + + base := t.TempDir() + helloDir := filepath.Join(base, "plugins", "hello") + if err := os.MkdirAll(helloDir, 0755); err != nil { + t.Fatal(err) + } + build := exec.Command(goBin, "build", "-o", filepath.Join(helloDir, "plugin"), "github.com/flatrun/agent/examples/plugins/hello") + if out, err := build.CombinedOutput(); err != nil { + t.Fatalf("building sample plugin: %v\n%s", err, out) + } + + h := New(filepath.Join(base, "plugins"), filepath.Join(base, "run"), "", "") + if err := h.Start(); err != nil { + t.Fatal(err) + } + defer h.Stop() + + infos := h.Infos() + if len(infos) != 1 || len(infos[0].Tools) != 1 || infos[0].Tools[0].Name != "echo" { + t.Fatalf("expected the echo tool advertised, got %+v", infos) + } + + out, err := h.ExecTool("hello", "echo", map[string]any{"text": "hi"}) + if err != nil { + t.Fatalf("ExecTool = %v", err) + } + if out != "echo: hi" { + t.Errorf("ExecTool result = %q, want %q", out, "echo: hi") + } + + if _, err := h.ExecTool("hello", "missing", nil); err == nil { + t.Error("expected error for unknown tool") + } +} + +// A built-in plugin launches from an explicit command (the agent re-execing itself), +// not from a scanned plugins directory. +func TestHostLaunchesBuiltinPlugin(t *testing.T) { + if testing.Short() { + t.Skip("skipping subprocess build in short mode") + } + goBin, err := exec.LookPath("go") + if err != nil { + t.Skip("go toolchain not available") + } + + base := t.TempDir() + bin := filepath.Join(base, "hello") + build := exec.Command(goBin, "build", "-o", bin, "github.com/flatrun/agent/examples/plugins/hello") + if out, err := build.CombinedOutput(); err != nil { + t.Fatalf("building sample plugin: %v\n%s", err, out) + } + + // No plugins dir; the plugin is registered as a built-in command. + h := New("", filepath.Join(base, "run"), "", "") + h.Builtin("hello", bin) + if err := h.Start(); err != nil { + t.Fatal(err) + } + defer h.Stop() + + proxy, ok := h.Proxy("hello") + if !ok { + t.Fatal("built-in plugin not running") + } + rec := httptest.NewRecorder() + proxy.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/hello", nil)) + if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "hello from the flatrun plugin") { + t.Fatalf("built-in proxied response = code %d body %q", rec.Code, rec.Body.String()) + } +} diff --git a/pkg/pluginapi/pluginapi.go b/pkg/pluginapi/pluginapi.go index c173a32..da192ac 100644 --- a/pkg/pluginapi/pluginapi.go +++ b/pkg/pluginapi/pluginapi.go @@ -12,6 +12,9 @@ const ( // EnvAgentURL and EnvToken let a plugin call back into the agent API. EnvAgentURL = "FLATRUN_AGENT_URL" EnvToken = "FLATRUN_PLUGIN_TOKEN" + // EnvDataDir is the deployments base directory, so a plugin can read/write flat state + // (e.g. its config under .flatrun/). + EnvDataDir = "FLATRUN_DATA_DIR" // InfoPath is the well-known endpoint every plugin serves. InfoPath = "/_plugin/info" @@ -19,11 +22,39 @@ const ( HandshakeHeader = "X-Flatrun-Plugin-Handshake" ) -// Info is a plugin's self-reported identity and the capabilities it requests from the host. +// Info is a plugin's self-reported identity, the capabilities it requests from the host, and +// how it contributes UI: which native slots it fills and any configuration it accepts. type Info struct { - Name string `json:"name"` - Version string `json:"version"` - DisplayName string `json:"display_name"` - Description string `json:"description"` - Capabilities []string `json:"capabilities,omitempty"` + Name string `json:"name"` + Version string `json:"version"` + DisplayName string `json:"display_name"` + Description string `json:"description"` + Capabilities []string `json:"capabilities,omitempty"` + UIExtensions []UIExtension `json:"ui_extensions,omitempty"` + ConfigSchema map[string]any `json:"config_schema,omitempty"` + Tools []ToolSpec `json:"tools,omitempty"` +} + +// ToolSpec declares a tool the plugin exposes to the AI assistant. The host advertises it to +// the model and dispatches calls back to the plugin's /_plugin/tools/exec endpoint. Mutates +// marks a tool that changes state, so the host can gate it on write access. +type ToolSpec struct { + Name string `json:"name"` + Description string `json:"description"` + Parameters map[string]any `json:"parameters,omitempty"` + Mutates bool `json:"mutates,omitempty"` +} + +// ToolExecPath is where the host POSTs {name, args} to invoke a plugin tool. +const ToolExecPath = "/_plugin/tools/exec" + +// UIExtension declares that a plugin contributes UI into a named slot. The host renders a +// native component for Kind (plugins never ship UI code) and feeds it from Endpoint, a path +// on the plugin's own API. +type UIExtension struct { + Slot string `json:"slot"` // "deployment.detail" | "settings" | "overview" + Kind string `json:"kind"` // "metrics-panel" | "form" | "timeline" + Title string `json:"title,omitempty"` + Icon string `json:"icon,omitempty"` + Endpoint string `json:"endpoint,omitempty"` } diff --git a/pkg/plugins/types.go b/pkg/plugins/types.go index cfb75a8..c7d1fc4 100644 --- a/pkg/plugins/types.go +++ b/pkg/plugins/types.go @@ -44,6 +44,7 @@ type PluginInfo struct { Requires []string `json:"requires,omitempty" yaml:"requires,omitempty"` Resources *ResourceRequirements `json:"resources,omitempty" yaml:"resources,omitempty"` DashboardExtensions []DashboardExtension `json:"dashboard_extensions,omitempty" yaml:"dashboard_extensions,omitempty"` + UIExtensions []UIExtension `json:"ui_extensions,omitempty" yaml:"ui_extensions,omitempty"` APIEndpoints []APIEndpoint `json:"api,omitempty" yaml:"api,omitempty"` Hooks map[string]string `json:"hooks,omitempty" yaml:"hooks,omitempty"` } @@ -53,6 +54,16 @@ type DashboardExtension struct { Component string `json:"component" yaml:"component"` } +// UIExtension declares a plugin's contribution to a named UI slot, rendered by a native +// component of the given Kind and fed from the plugin's own API Endpoint. +type UIExtension struct { + Slot string `json:"slot" yaml:"slot"` + Kind string `json:"kind" yaml:"kind"` + Title string `json:"title,omitempty" yaml:"title,omitempty"` + Icon string `json:"icon,omitempty" yaml:"icon,omitempty"` + Endpoint string `json:"endpoint,omitempty" yaml:"endpoint,omitempty"` +} + type APIEndpoint struct { Path string `json:"path" yaml:"path"` Method string `json:"method" yaml:"method"` diff --git a/pkg/pluginsdk/sdk.go b/pkg/pluginsdk/sdk.go index fdc821e..9705017 100644 --- a/pkg/pluginsdk/sdk.go +++ b/pkg/pluginsdk/sdk.go @@ -16,22 +16,58 @@ import ( "github.com/flatrun/agent/pkg/pluginapi" ) -// Serve runs the plugin until the host stops it. handler serves the plugin's own routes; -// pass nil if the plugin only reports info. It returns an error if the process was not -// launched by the host (the socket env var is unset) or the socket cannot be served. -func Serve(info pluginapi.Info, handler http.Handler) error { +// Tool is a plugin capability the AI assistant can invoke. Run receives the parsed args and +// returns the textual result the model reads. +type Tool struct { + Spec pluginapi.ToolSpec + Run func(args map[string]any) (string, error) +} + +// Serve runs the plugin until the host stops it. handler serves the plugin's own routes (nil +// if the plugin only reports info); tools are advertised to the assistant and dispatched over +// the tool-exec endpoint. It returns an error if the process was not launched by the host +// (the socket env var is unset) or the socket cannot be served. +func Serve(info pluginapi.Info, handler http.Handler, tools ...Tool) error { socket := os.Getenv(pluginapi.EnvSocket) if socket == "" { return fmt.Errorf("not launched by the flatrun plugin host: %s is unset", pluginapi.EnvSocket) } handshake := os.Getenv(pluginapi.EnvHandshake) + byName := make(map[string]Tool, len(tools)) + for _, t := range tools { + byName[t.Spec.Name] = t + info.Tools = append(info.Tools, t.Spec) + } + mux := http.NewServeMux() mux.HandleFunc(pluginapi.InfoPath, func(w http.ResponseWriter, _ *http.Request) { w.Header().Set(pluginapi.HandshakeHeader, handshake) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(info) }) + mux.HandleFunc(pluginapi.ToolExecPath, func(w http.ResponseWriter, r *http.Request) { + var req struct { + Name string `json:"name"` + Args map[string]any `json:"args"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + tool, ok := byName[req.Name] + if !ok { + http.Error(w, "unknown tool", http.StatusNotFound) + return + } + result, err := tool.Run(req.Args) + w.Header().Set("Content-Type", "application/json") + if err != nil { + _ = json.NewEncoder(w).Encode(map[string]string{"error": err.Error()}) + return + } + _ = json.NewEncoder(w).Encode(map[string]string{"result": result}) + }) if handler != nil { mux.Handle("/", handler) }