diff --git a/Makefile b/Makefile index 86ddb4f..9dde67e 100644 --- a/Makefile +++ b/Makefile @@ -2,6 +2,26 @@ BINARY := bin/flow +# --- env defaults for `make watch` ---------------------------------------- +# These match the host-side ports published by docker-compose.yml. Override +# any value at the command line, e.g. `make watch MINIO_ENDPOINT=...`. +# +# Anything you `export` separately in your shell still takes precedence over +# these, since the Go binary only reads its env via os.Getenv. +export FLOW_ADDR ?= :8090 +export FLOW_CORS_ORIGINS ?= http://localhost:3000 +export MONGO_URI ?= mongodb://localhost:27017 +export MONGO_DB ?= flow +export RESTATE_INGRESS_URL ?= http://localhost:8081 +export RESTATE_ADMIN_URL ?= http://localhost:9070 +export RESTATE_DEPLOYMENT_URI ?= http://host.docker.internal:9080 +export BUILDKIT_HOST ?= tcp://127.0.0.1:1234 +export MINIO_ENDPOINT ?= 127.0.0.1:9000 +export MINIO_ACCESS_KEY ?= minio +export MINIO_SECRET_KEY ?= minio12345 +export MINIO_BUCKET ?= flow-logs +export MINIO_USE_SSL ?= false + build: web go build -o $(BINARY) ./cmd/flow @@ -16,7 +36,7 @@ web-dev: cd web && (test -d node_modules || npm install --no-fund --no-audit --loglevel=error) && npm run dev # `make dev` runs the Next.js dev server (auto-installs deps). -# In another terminal run `make watch` to hot-reload the Go API on :8080; +# In another terminal run `make watch` to hot-reload the Go API on :8090; # the Next dev rewrite forwards /api/* requests to it. dev: web-dev @@ -24,12 +44,20 @@ serve: build-go ./$(BINARY) serve # Hot-reload the Go server with air. Re-run on every *.go change. +# Picks up MINIO_*, MONGO_*, RESTATE_*, BUILDKIT_HOST, FLOW_* defaults from +# this Makefile so a fresh shell can `make watch` without exporting anything. # First time: `go install github.com/air-verse/air@latest` watch: @command -v air >/dev/null 2>&1 || { \ echo "installing air..."; \ go install github.com/air-verse/air@latest; \ } + @echo "flow env:" + @echo " FLOW_ADDR = $(FLOW_ADDR)" + @echo " MONGO_URI = $(MONGO_URI)" + @echo " RESTATE_INGRESS_URL = $(RESTATE_INGRESS_URL)" + @echo " BUILDKIT_HOST = $(BUILDKIT_HOST)" + @echo " MINIO_ENDPOINT = $(MINIO_ENDPOINT)" air run: build diff --git a/README.md b/README.md index 90b8ef3..7620463 100644 --- a/README.md +++ b/README.md @@ -50,26 +50,20 @@ Pre-v0.1. Working but moving fast — APIs and node types may change. ## Quickstart -The default `docker compose up` brings the **API + UI + backing stores** -(Mongo, Restate). It assumes you have BuildKit + a registry running on -the host already (e.g. via the sibling `langship` stack). - ```sh docker compose up # UI: http://localhost:3000 # API: http://localhost:8090 # Restate: :8081 ingress, :9070 admin +# BuildKit: 127.0.0.1:1234 +# Registry: 127.0.0.1:5050 (host port; buildkitd pushes to registry:5000 internally) +# MinIO: 127.0.0.1:9000 (S3 API), :9001 (console; minio / minio12345) ``` -If you **don't** have BuildKit + registry running, layer the standalone -overlay to bring them up too: - -```sh -docker compose -f docker-compose.yml -f docker-compose.standalone.yml up -# adds: -# buildkitd 127.0.0.1:1234 (moby/buildkit:v0.18.2) -# registry 127.0.0.1:5000 (registry:2) -``` +The base compose now bundles every service flow needs: **mongo, restate, +buildkitd, registry, minio, flow, web**. If a sibling stack already owns +one of those host ports (e.g. another langship-* set), stop that +container or override the port mapping in a `compose.override.yml`. ## Dev (hot reload) diff --git a/cmd/flow/main.go b/cmd/flow/main.go index 748bcb0..7efac0c 100644 --- a/cmd/flow/main.go +++ b/cmd/flow/main.go @@ -17,6 +17,7 @@ import ( "github.com/lyzrai/flow/pkg/engine" "github.com/lyzrai/flow/pkg/execevents" "github.com/lyzrai/flow/pkg/executors" + "github.com/lyzrai/flow/pkg/logstore" "github.com/lyzrai/flow/pkg/orchestrator" "github.com/lyzrai/flow/pkg/storage" ) @@ -132,6 +133,33 @@ func serve() int { }() slog.Info("mongo connected", slog.String("db", mongoDB)) + // MinIO is optional — when MINIO_ENDPOINT is unset we skip log + // archiving. Live SSE log streaming still works regardless. + var logs logstore.Store + if endpoint := envOr("MINIO_ENDPOINT", ""); endpoint != "" { + minioCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + m, err := logstore.NewMinio(minioCtx, logstore.Config{ + Endpoint: endpoint, + AccessKey: envOr("MINIO_ACCESS_KEY", "minio"), + SecretKey: envOr("MINIO_SECRET_KEY", "minio12345"), + Bucket: envOr("MINIO_BUCKET", "flow-logs"), + UseSSL: envOr("MINIO_USE_SSL", "false") == "true", + }) + cancel() + if err != nil { + slog.Warn("minio init failed, log archive disabled", + slog.String("endpoint", endpoint), + slog.Any("error", err), + ) + } else { + logs = m + slog.Info("minio connected", + slog.String("endpoint", endpoint), + slog.String("bucket", envOr("MINIO_BUCKET", "flow-logs")), + ) + } + } + // Register executors now that storage is ready; the Build executor // reads from AgentStore. executors.RegisterAll(executors.RegistryDeps{ @@ -189,6 +217,7 @@ func serve() int { Runs: mongo.Runs(), Agents: mongo.Agents(), Events: eventBus, + Logs: logs, }), ReadHeaderTimeout: 10 * time.Second, } diff --git a/docker-compose.standalone.yml b/docker-compose.standalone.yml deleted file mode 100644 index 537d80d..0000000 --- a/docker-compose.standalone.yml +++ /dev/null @@ -1,55 +0,0 @@ -# Standalone overlay — adds buildkitd + registry to the base -# docker-compose.yml when you DON'T have the sibling `langship` stack -# already providing them on the host (1234 + 5000). -# -# Use: -# docker compose -f docker-compose.yml -f docker-compose.standalone.yml up -# -# This re-binds `flow` to point BUILDKIT_HOST at the in-network buildkitd -# instead of host.docker.internal, and adds explicit dependencies. - -services: - registry: - image: registry:2 - container_name: langship-flow-registry - ports: - - "127.0.0.1:5000:5000" - volumes: - - registry-data:/var/lib/registry - - buildkitd: - image: moby/buildkit:v0.18.2 - container_name: langship-flow-buildkitd - privileged: true - command: - - --addr - - tcp://0.0.0.0:1234 - - --addr - - unix:///run/buildkit/buildkitd.sock - - --config - - /etc/buildkit/buildkitd.toml - ports: - - "127.0.0.1:1234:1234" - volumes: - - buildkit-data:/var/lib/buildkit - - ./docker/buildkit/buildkitd.toml:/etc/buildkit/buildkitd.toml:ro - healthcheck: - test: ["CMD", "buildctl", "--addr", "tcp://0.0.0.0:1234", "debug", "info"] - interval: 10s - timeout: 5s - retries: 10 - start_period: 10s - - flow: - environment: - # In standalone mode, buildkitd is a sibling on the same compose network. - - BUILDKIT_HOST=tcp://buildkitd:1234 - depends_on: - - mongo - - restate - - buildkitd - - registry - -volumes: - buildkit-data: - registry-data: diff --git a/docker-compose.yml b/docker-compose.yml index 5121698..902a4e5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -12,12 +12,57 @@ services: - "8081:8080" # ingress - "9070:9070" # admin - # NOTE: this stack reuses the buildkitd + registry from the sibling - # `langship` compose project — both already publish on - # 127.0.0.1:1234 (BuildKit) and 127.0.0.1:5000 (registry). Bringing them - # up here would collide on the same host ports. If you ever run flow - # standalone (without that other stack), uncomment the services in - # docker-compose.standalone.yml. + registry: + image: registry:2 + container_name: langship-flow-registry + ports: + # Host port 5050 (debug / docker pull from host); inside the compose + # network buildkitd pushes to registry:5000 via service DNS. + - "127.0.0.1:5050:5000" + volumes: + - registry-data:/var/lib/registry + + buildkitd: + image: moby/buildkit:v0.18.2 + container_name: langship-flow-buildkitd + privileged: true + command: + - --addr + - tcp://0.0.0.0:1234 + - --addr + - unix:///run/buildkit/buildkitd.sock + - --config + - /etc/buildkit/buildkitd.toml + ports: + - "127.0.0.1:1234:1234" + volumes: + - buildkit-data:/var/lib/buildkit + - ./docker/buildkit/buildkitd.toml:/etc/buildkit/buildkitd.toml:ro + healthcheck: + test: ["CMD", "buildctl", "--addr", "tcp://0.0.0.0:1234", "debug", "info"] + interval: 10s + timeout: 5s + retries: 10 + start_period: 10s + + minio: + image: minio/minio:latest + container_name: langship-flow-minio + command: server /data --console-address ":9001" + environment: + - MINIO_ROOT_USER=minio + - MINIO_ROOT_PASSWORD=minio12345 + ports: + - "127.0.0.1:9000:9000" # S3 API + - "127.0.0.1:9001:9001" # web console + volumes: + - minio-data:/data + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"] + interval: 10s + timeout: 3s + retries: 5 + start_period: 5s flow: build: . @@ -32,14 +77,18 @@ services: - RESTATE_INGRESS_URL=http://restate:8080 - RESTATE_ADMIN_URL=http://restate:9070 - RESTATE_DEPLOYMENT_URI=http://flow:9080 - - BUILDKIT_HOST=tcp://host.docker.internal:1234 + - BUILDKIT_HOST=tcp://buildkitd:1234 + - MINIO_ENDPOINT=minio:9000 + - MINIO_ACCESS_KEY=minio + - MINIO_SECRET_KEY=minio12345 + - MINIO_BUCKET=flow-logs + - MINIO_USE_SSL=false depends_on: - mongo - restate - # buildkitd + registry are external (the sibling langship stack); see - # the note above. Add `external_links: ["langship-buildkitd:buildkitd", - # "langship-registry:registry"]` if you run flow as docker-compose and - # need network connectivity to those containers. + - buildkitd + - registry + - minio web: build: ./web @@ -50,3 +99,6 @@ services: volumes: mongo-data: + buildkit-data: + registry-data: + minio-data: diff --git a/go.mod b/go.mod index 055a2ff..6073682 100644 --- a/go.mod +++ b/go.mod @@ -1,15 +1,16 @@ module github.com/lyzrai/flow -go 1.24.5 +go 1.25 require ( github.com/docker/cli v27.4.0-rc.2+incompatible github.com/dop251/goja v0.0.0-20260311135729-065cd970411c github.com/google/uuid v1.6.0 + github.com/minio/minio-go/v7 v7.1.0 github.com/moby/buildkit v0.18.2 github.com/restatedev/sdk-go v0.23.0 go.mongodb.org/mongo-driver/v2 v2.6.0 - golang.org/x/sync v0.18.0 + golang.org/x/sync v0.19.0 ) require ( @@ -17,6 +18,7 @@ require ( github.com/Microsoft/go-winio v0.6.2 // indirect github.com/bahlo/generic-list-go v0.2.0 // indirect github.com/buger/jsonparser v1.1.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/containerd/console v1.0.4 // indirect github.com/containerd/containerd v1.7.24 // indirect github.com/containerd/containerd/api v1.8.0 // indirect @@ -30,7 +32,9 @@ require ( github.com/dlclark/regexp2 v1.11.4 // indirect github.com/docker/docker-credential-helpers v0.8.2 // indirect github.com/docker/go-units v0.5.0 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect + github.com/go-ini/ini v1.67.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-sourcemap/sourcemap v2.1.3+incompatible // indirect @@ -47,8 +51,12 @@ require ( github.com/hashicorp/go-multierror v1.1.1 // indirect github.com/in-toto/in-toto-golang v0.5.0 // indirect github.com/invopop/jsonschema v0.13.0 // indirect - github.com/klauspost/compress v1.18.0 // indirect + github.com/klauspost/compress v1.18.2 // indirect + github.com/klauspost/cpuid/v2 v2.2.11 // indirect + github.com/klauspost/crc32 v1.3.0 // indirect github.com/mailru/easyjson v0.7.7 // indirect + github.com/minio/crc64nvme v1.1.1 // indirect + github.com/minio/md5-simd v1.1.2 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect github.com/moby/locker v1.0.1 // indirect github.com/moby/patternmatcher v0.6.0 // indirect @@ -57,12 +65,15 @@ require ( github.com/mr-tron/base58 v1.2.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/philhofer/fwd v1.2.0 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect + github.com/rs/xid v1.6.0 // indirect github.com/secure-systems-lab/go-securesystemslib v0.4.0 // indirect github.com/shibumi/go-pathspec v1.3.0 // indirect github.com/sirupsen/logrus v1.9.3 // indirect github.com/tetratelabs/wazero v1.9.0 // indirect + github.com/tinylib/msgp v1.6.1 // indirect github.com/tonistiigi/fsutil v0.0.0-20241121093142-31cf1f437184 // indirect github.com/tonistiigi/go-csvvalue v0.0.0-20240710180619-ddb21b71c0b4 // indirect github.com/tonistiigi/units v0.0.0-20180711220420-6950e57a87ea // indirect @@ -72,6 +83,7 @@ require ( github.com/xdg-go/scram v1.2.0 // indirect github.com/xdg-go/stringprep v1.0.4 // indirect github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect + github.com/zeebo/xxh3 v1.1.0 // indirect go.opentelemetry.io/auto/sdk v1.1.0 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.46.1 // indirect @@ -82,10 +94,11 @@ require ( go.opentelemetry.io/otel/sdk v1.38.0 // indirect go.opentelemetry.io/otel/trace v1.38.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect - golang.org/x/crypto v0.45.0 // indirect - golang.org/x/net v0.47.0 // indirect - golang.org/x/sys v0.38.0 // indirect - golang.org/x/text v0.31.0 // indirect + go.yaml.in/yaml/v3 v3.0.4 // indirect + golang.org/x/crypto v0.46.0 // indirect + golang.org/x/net v0.48.0 // indirect + golang.org/x/sys v0.39.0 // indirect + golang.org/x/text v0.32.0 // indirect golang.org/x/time v0.12.0 // indirect google.golang.org/genproto v0.0.0-20240123012728-ef4313101c80 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 // indirect diff --git a/go.sum b/go.sum index 199d1f0..336cbe4 100644 --- a/go.sum +++ b/go.sum @@ -18,6 +18,8 @@ github.com/bahlo/generic-list-go v0.2.0 h1:5sz/EEAK+ls5wF+NeqDpk5+iNdMDXrh3z3nPn github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xWbdbCW3pNTGyYg= github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs= github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cncf/xds/go v0.0.0-20250501225837-2ac532fd4443 h1:aQ3y1lwWyqYPiWZThqv1aFbZMiM9vblcSArJRf2Irls= github.com/cncf/xds/go v0.0.0-20250501225837-2ac532fd4443/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8= github.com/codahale/rfc6979 v0.0.0-20141003034818-6a90f24967eb h1:EDmT6Q9Zs+SbUoc7Ik9EfrFqcylYqgPZ9ANSbTAntnE= @@ -73,10 +75,14 @@ github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= github.com/dop251/goja v0.0.0-20260311135729-065cd970411c h1:OcLmPfx1T1RmZVHHFwWMPaZDdRf0DBMZOFMVWJa7Pdk= github.com/dop251/goja v0.0.0-20260311135729-065cd970411c/go.mod h1:MxLav0peU43GgvwVgNbLAj1s/bSGboKkhuULvq/7hx4= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/envoyproxy/protoc-gen-validate v1.2.1 h1:DEo3O99U8j4hBFwbJfrz9VtgcDfUKS7KJ7spH3d86P8= github.com/envoyproxy/protoc-gen-validate v1.2.1/go.mod h1:d/C80l/jxXLdfEIhX1W2TmLfsJ31lvEjwamM4DxlWXU= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/go-ini/ini v1.67.0 h1:z6ZrTEZqSWOTyH2FlglNbNgARyHG8oLW9gMELqKr06A= +github.com/go-ini/ini v1.67.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= @@ -118,14 +124,25 @@ github.com/invopop/jsonschema v0.13.0/go.mod h1:ffZ5Km5SWWRAIN6wbDXItl95euhFz2uO github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= -github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= -github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk= +github.com/klauspost/compress v1.18.2/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= +github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= +github.com/klauspost/cpuid/v2 v2.2.11 h1:0OwqZRYI2rFrjS4kvkDnqJkKHdHaRnCm68/DY4OxRzU= +github.com/klauspost/cpuid/v2 v2.2.11/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/klauspost/crc32 v1.3.0 h1:sSmTt3gUt81RP655XGZPElI0PelVTZ6YwCRnPSupoFM= +github.com/klauspost/crc32 v1.3.0/go.mod h1:D7kQaZhnkX/Y0tstFGf8VUzv2UofNGqCjnC3zdHB0Hw= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= +github.com/minio/crc64nvme v1.1.1 h1:8dwx/Pz49suywbO+auHCBpCtlW1OfpcLN7wYgVR6wAI= +github.com/minio/crc64nvme v1.1.1/go.mod h1:eVfm2fAzLlxMdUGc0EEBGSMmPwmXD5XiNRpnu9J3bvg= +github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34= +github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM= +github.com/minio/minio-go/v7 v7.1.0 h1:QEt5IStDpxgGjEdtOgpiZ5QhmSl3ax7qy61vi2SwHO8= +github.com/minio/minio-go/v7 v7.1.0/go.mod h1:Dm7WS1AgLmBa0NcQD6SeJnJf+K/EUW3GR7Ks6olB3OA= github.com/moby/buildkit v0.18.2 h1:l86uBvxh4ntNoUUg3Y0eGTbKg1PbUh6tawJ4Xt75SpQ= github.com/moby/buildkit v0.18.2/go.mod h1:vCR5CX8NGsPTthTg681+9kdmfvkvqJBXEv71GZe5msU= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= @@ -160,6 +177,8 @@ github.com/opencontainers/selinux v1.11.0 h1:+5Zbo97w3Lbmb3PeqQtpmTkMwsW5nRI3YaL github.com/opencontainers/selinux v1.11.0/go.mod h1:E5dMC3VPuVvVHDYmi78qvhJp8+M586T4DlDRYpFkyec= 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/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM= +github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= 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= @@ -172,6 +191,8 @@ github.com/restatedev/sdk-go v0.23.0 h1:Eewh6n/YZUfA7In5ZiAIA6smLyssynsBHVijUguN github.com/restatedev/sdk-go v0.23.0/go.mod h1:2G757yGe0Ihwcb+Z/HZUscQ0g3PFTyueO0f8qlqxWDo= github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= +github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU= +github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= github.com/secure-systems-lab/go-securesystemslib v0.4.0 h1:b23VGrQhTA8cN2CbBw7/FulN9fTtqYUdS5+Oxzt+DUE= github.com/secure-systems-lab/go-securesystemslib v0.4.0/go.mod h1:FGBZgq2tXWICsxWQW1msNf49F0Pf2Op5Htayx335Qbs= github.com/shibumi/go-pathspec v1.3.0 h1:QUyMZhFo0Md5B8zV8x2tesohbb5kfbpTi9rBnKh5dkI= @@ -187,6 +208,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/tetratelabs/wazero v1.9.0 h1:IcZ56OuxrtaEz8UYNRHBrUa9bYeX9oVY93KspZZBf/I= github.com/tetratelabs/wazero v1.9.0/go.mod h1:TSbcXCfFP0L2FGkRPxHphadXPjo1T6W+CseNNY7EkjM= +github.com/tinylib/msgp v1.6.1 h1:ESRv8eL3u+DNHUoSAAQRE50Hm162zqAnBoGv9PzScPY= +github.com/tinylib/msgp v1.6.1/go.mod h1:RSp0LW9oSxFut3KzESt5Voq4GVWyS+PSulT77roAqEA= github.com/tonistiigi/fsutil v0.0.0-20241121093142-31cf1f437184 h1:RgyoSI38Y36zjQaszel/0RAcIehAnjA1B0RiUV9SDO4= github.com/tonistiigi/fsutil v0.0.0-20241121093142-31cf1f437184/go.mod h1:Dl/9oEjK7IqnjAm21Okx/XIxUCFJzvh+XdVHUlBwXTw= github.com/tonistiigi/go-csvvalue v0.0.0-20240710180619-ddb21b71c0b4 h1:7I5c2Ig/5FgqkYOh/N87NzoyI9U15qUPXhDD8uCupv8= @@ -210,6 +233,10 @@ github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfS github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ= +github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.mongodb.org/mongo-driver/v2 v2.6.0 h1:b9sJOYrkmt4l8bY43ZenFBcPlhYIjaOfYHLtbB/5qi8= go.mongodb.org/mongo-driver/v2 v2.6.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0= @@ -238,33 +265,35 @@ go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjce go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= 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.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= -golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q= -golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4= +golang.org/x/crypto v0.46.0 h1:cKRW/pmt1pKAfetfu+RCEvjvZkA9RimPbh7bhFjGVBU= +golang.org/x/crypto v0.46.0/go.mod h1:Evb/oLKmMraqjZ2iQTwDwvCtJkczlDuTmdJXoZVzqU0= 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.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= -golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA= -golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w= +golang.org/x/mod v0.30.0 h1:fDEXFVZ/fmCKProc/yAXXUijritrDzahmwwefnjoPFk= +golang.org/x/mod v0.30.0/go.mod h1:lAsf5O2EvJeSFMiBxXDki7sCgAxEUcZHXoXMKT4GJKc= 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.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= -golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= -golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= +golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU= +golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= golang.org/x/oauth2 v0.30.0 h1:dnDm7JmhM45NNpd8FDDeLhK6FwqbOf4MLCM9zb1BOHI= golang.org/x/oauth2 v0.30.0/go.mod h1:B++QgG3ZKulg6sRPGD/mqlHQs5rB3Ml9erfeDY7xKlU= 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.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= -golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +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/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-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -274,18 +303,18 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= -golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk= +golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= -golang.org/x/term v0.37.0 h1:8EGAD0qCmHYZg6J17DvsMy9/wJ7/D/4pV/wfnld5lTU= -golang.org/x/term v0.37.0/go.mod h1:5pB4lxRNYYVZuTLmy8oR2BH8dflOR+IbTYFD8fi3254= +golang.org/x/term v0.38.0 h1:PQ5pkm/rLO6HnxFR7N2lJHOZX6Kez5Y1gDSJla6jo7Q= +golang.org/x/term v0.38.0/go.mod h1:bSEAKrOT1W+VSu9TSCMtoGEOUcKxOKgl3LE5QEF/xVg= 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.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= -golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= -golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= +golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU= +golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY= golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= diff --git a/pkg/api/agents.go b/pkg/api/agents.go index 4c28f1d..2de7052 100644 --- a/pkg/api/agents.go +++ b/pkg/api/agents.go @@ -434,6 +434,28 @@ func (s *Server) dispatchAgent(ctx context.Context, a *storage.Agent, trigger an }) continue } + // Hook the log archiver onto the new execution so per-node lines + // land in MinIO when each node finishes. + startLogArchiver(context.Background(), s.logs, s.events, execID) + + // Broadcast so /runs etc. light up without polling. We extract + // `source` from the trigger payload (manual / github_push). + source := "" + if items, ok := trigger.([]map[string]any); ok && len(items) > 0 { + if s, _ := items[0]["source"].(string); s != "" { + source = s + } + } + s.runsBus.Publish(RunCreatedEvent{ + Type: "run_created", + ExecutionID: execID, + PipelineID: pid, + PipelineName: p.Name, + AgentID: a.ID, + Source: source, + StartedAt: time.Now().UTC(), + }) + if s.runs != nil { _ = s.runs.Insert(ctx, &storage.Run{ ID: execID, diff --git a/pkg/api/log_archiver.go b/pkg/api/log_archiver.go new file mode 100644 index 0000000..1bcae2a --- /dev/null +++ b/pkg/api/log_archiver.go @@ -0,0 +1,140 @@ +package api + +import ( + "bytes" + "context" + "log/slog" + "strings" + "sync" + "time" + + "github.com/lyzrai/flow/pkg/engine" + "github.com/lyzrai/flow/pkg/logstore" +) + +// logArchiver subscribes to the in-memory event bus for a single execution, +// buffers `node_log` lines per node, and flushes each node's buffer to the +// log store when the node finishes (node_completed / node_error). On the +// terminal `done` event it flushes any leftovers and stops. +// +// One archiver per active execution. They drop themselves from the +// archive map when finished. +type logArchiver struct { + execID string + store logstore.Store + bus EventSubscriber + + // in-memory buffer per node — the live SSE listener (the UI) reads from + // the bus directly; this struct is purely for archive-on-finish. + mu sync.Mutex + buffers map[string]*bytes.Buffer +} + +// startLogArchiver spins up a goroutine that drains the per-exec event bus +// into MinIO. Safe to call once per execution; idempotent if the bus is nil. +func startLogArchiver(parentCtx context.Context, store logstore.Store, bus EventSubscriber, execID string) { + if store == nil || bus == nil || execID == "" { + return + } + ar := &logArchiver{ + execID: execID, + store: store, + bus: bus, + buffers: map[string]*bytes.Buffer{}, + } + go ar.run(parentCtx) +} + +func (a *logArchiver) run(parentCtx context.Context) { + ch, cancel := a.bus.Subscribe(a.execID) + defer cancel() + + // Detach from the request context that triggered the run; once submitted + // we want to keep archiving even if the caller disconnects. Cap with a + // per-run deadline so a stuck workflow doesn't leak this goroutine + // forever. + ctx, cancelCtx := context.WithTimeout(context.Background(), 1*time.Hour) + defer cancelCtx() + + for { + select { + case <-ctx.Done(): + a.flushAll(ctx) + return + case <-parentCtx.Done(): + // Process is shutting down. + a.flushAll(context.Background()) + return + case ev, ok := <-ch: + if !ok { + a.flushAll(ctx) + return + } + a.handle(ctx, ev) + if ev.Type == engine.EventDone { + a.flushAll(ctx) + return + } + } + } +} + +func (a *logArchiver) handle(ctx context.Context, ev engine.ExecutionEvent) { + switch ev.Type { + case engine.EventNodeLog: + if ev.Node == "" || ev.Content == "" { + return + } + a.mu.Lock() + buf := a.buffers[ev.Node] + if buf == nil { + buf = &bytes.Buffer{} + a.buffers[ev.Node] = buf + } + // One line per Log event; force a trailing newline so the archived + // file is line-oriented and easy to tail. + buf.WriteString(strings.TrimRight(ev.Content, "\r\n")) + buf.WriteByte('\n') + a.mu.Unlock() + + case engine.EventNodeCompleted, engine.EventNodeError: + if ev.Node == "" { + return + } + a.flushNode(ctx, ev.Node) + } +} + +func (a *logArchiver) flushNode(ctx context.Context, node string) { + a.mu.Lock() + buf := a.buffers[node] + if buf == nil || buf.Len() == 0 { + a.mu.Unlock() + return + } + data := append([]byte(nil), buf.Bytes()...) + delete(a.buffers, node) + a.mu.Unlock() + + putCtx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + if err := a.store.Put(putCtx, a.execID, node, data); err != nil { + slog.WarnContext(ctx, "log_archive_failed", + slog.String("execution_id", a.execID), + slog.String("node", node), + slog.Any("error", err), + ) + } +} + +func (a *logArchiver) flushAll(ctx context.Context) { + a.mu.Lock() + nodes := make([]string, 0, len(a.buffers)) + for n := range a.buffers { + nodes = append(nodes, n) + } + a.mu.Unlock() + for _, n := range nodes { + a.flushNode(ctx, n) + } +} diff --git a/pkg/api/runs_bus.go b/pkg/api/runs_bus.go new file mode 100644 index 0000000..753b55a --- /dev/null +++ b/pkg/api/runs_bus.go @@ -0,0 +1,71 @@ +package api + +import ( + "encoding/json" + "sync" + "time" +) + +// RunCreatedEvent is the payload broadcast on /api/runs/stream whenever a +// new run is dispatched (manual API call or GitHub webhook). +type RunCreatedEvent struct { + Type string `json:"type"` // "run_created" + ExecutionID string `json:"executionId"` + PipelineID string `json:"pipelineId,omitempty"` + PipelineName string `json:"pipelineName,omitempty"` + AgentID string `json:"agentId,omitempty"` + Source string `json:"source,omitempty"` // "manual" | "github_push" + StartedAt time.Time `json:"startedAt"` +} + +// runsBus is a tiny broadcast bus for cross-execution UI events. The SSE +// stream handler subscribes; dispatchers publish. All in-memory; one bus +// per process. +type runsBus struct { + mu sync.RWMutex + subscribers map[chan RunCreatedEvent]struct{} +} + +func newRunsBus() *runsBus { + return &runsBus{subscribers: map[chan RunCreatedEvent]struct{}{}} +} + +// Publish fans an event out to every active subscriber. Non-blocking — slow +// subscribers drop the event so dispatchers never stall. +func (b *runsBus) Publish(ev RunCreatedEvent) { + b.mu.RLock() + subs := make([]chan RunCreatedEvent, 0, len(b.subscribers)) + for c := range b.subscribers { + subs = append(subs, c) + } + b.mu.RUnlock() + for _, c := range subs { + select { + case c <- ev: + default: + } + } +} + +// Subscribe registers a buffered channel and returns it + an unsubscribe +// func that closes the channel exactly once. +func (b *runsBus) Subscribe() (<-chan RunCreatedEvent, func()) { + ch := make(chan RunCreatedEvent, 16) + b.mu.Lock() + b.subscribers[ch] = struct{}{} + b.mu.Unlock() + var once sync.Once + cancel := func() { + once.Do(func() { + b.mu.Lock() + delete(b.subscribers, ch) + b.mu.Unlock() + close(ch) + }) + } + return ch, cancel +} + +// marshal is a small helper so handlers don't import encoding/json just for +// the event payload format. +func (e RunCreatedEvent) marshal() ([]byte, error) { return json.Marshal(e) } diff --git a/pkg/api/server.go b/pkg/api/server.go index c228d52..65fdbfb 100644 --- a/pkg/api/server.go +++ b/pkg/api/server.go @@ -21,6 +21,7 @@ import ( "time" "github.com/lyzrai/flow/pkg/engine" + "github.com/lyzrai/flow/pkg/logstore" "github.com/lyzrai/flow/pkg/models" "github.com/lyzrai/flow/pkg/orchestrator" "github.com/lyzrai/flow/pkg/storage" @@ -56,6 +57,11 @@ type ServerDeps struct { // per-node lifecycle events to. The SSE handler subscribes per // execution ID. Nil disables /api/executions/{id}/stream. Events EventSubscriber + + // Logs is the archive backend (MinIO/S3). When set, every dispatched + // run starts a background archiver that flushes per-node log buffers + // to object storage. Nil disables archiving (live SSE still works). + Logs logstore.Store } // EventSubscriber is the slice of execevents.MemoryBus the API needs. @@ -77,6 +83,12 @@ type Server struct { runs storage.RunStore agents storage.AgentStore events EventSubscriber + logs logstore.Store + + // runsBus broadcasts run_created events to every UI tab subscribed to + // /api/runs/stream. Used so a webhook-triggered run shows up live in + // the dashboard / runs list / agent detail page without polling. + runsBus *runsBus } // NewServer constructs an API-only Server. deps.Orchestrator may be nil — @@ -93,6 +105,8 @@ func NewServer(deps ServerDeps) *Server { runs: deps.Runs, agents: deps.Agents, events: deps.Events, + logs: deps.Logs, + runsBus: newRunsBus(), } s.routes() return s @@ -119,8 +133,10 @@ func (s *Server) routes() { s.mux.HandleFunc("DELETE /api/workflows/{id}", s.handleDeleteFlow) s.mux.HandleFunc("POST /api/workflows/execute", s.handleExecuteWorkflow) s.mux.HandleFunc("GET /api/executions", s.handleListExecutions) + s.mux.HandleFunc("GET /api/runs/stream", s.handleRunsStream) s.mux.HandleFunc("GET /api/executions/{id}", s.handleGetExecution) s.mux.HandleFunc("GET /api/executions/{id}/stream", s.handleStreamExecution) + s.mux.HandleFunc("GET /api/executions/{id}/logs/{node}", s.handleNodeLog) s.mux.HandleFunc("POST /api/executions/{id}/resume", s.handleResumeExecution) // Agents — Langship-style agent registry (git URL + PAT) @@ -362,6 +378,21 @@ func (s *Server) handleExecuteWorkflow(w http.ResponseWriter, r *http.Request) { return } + // Subscribe the log archiver to this execution. Drains node_log events + // off the event bus and flushes per-node buffers to MinIO when each + // node completes. + startLogArchiver(context.Background(), s.logs, s.events, execID) + + // Broadcast so /runs and other tabs flip to live without polling. + s.runsBus.Publish(RunCreatedEvent{ + Type: "run_created", + ExecutionID: execID, + PipelineID: pipelineID, + PipelineName: pipelineName, + Source: "manual", + StartedAt: time.Now().UTC(), + }) + // Best-effort run record. A failure here shouldn't block the response — // the orchestrator already accepted the workflow. if s.runs != nil { @@ -545,6 +576,83 @@ func (s *Server) handleStreamExecution(w http.ResponseWriter, r *http.Request) { } } +// handleRunsStream is a global SSE feed of run_created events. UI tabs +// subscribe once and react to runs from any source (manual trigger, agent +// trigger, GitHub push). Heartbeats every 15s. +func (s *Server) handleRunsStream(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + writeError(w, http.StatusInternalServerError, errors.New("streaming not supported")) + return + } + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) + + ch, cancel := s.runsBus.Subscribe() + defer cancel() + + _, _ = fmt.Fprint(w, "event: open\ndata: {}\n\n") + flusher.Flush() + + heartbeat := time.NewTicker(15 * time.Second) + defer heartbeat.Stop() + + for { + select { + case <-r.Context().Done(): + return + case <-heartbeat.C: + _, _ = fmt.Fprint(w, ": heartbeat\n\n") + flusher.Flush() + case ev, ok := <-ch: + if !ok { + return + } + payload, err := ev.marshal() + if err != nil { + continue + } + _, _ = fmt.Fprintf(w, "data: %s\n\n", payload) + flusher.Flush() + } + } +} + +// handleNodeLog streams the archived log for one node of an execution. +// Reads through the configured logstore (MinIO/S3 in prod) so the browser +// never talks to the object store directly. text/plain. +func (s *Server) handleNodeLog(w http.ResponseWriter, r *http.Request) { + if s.logs == nil { + writeError(w, http.StatusServiceUnavailable, errors.New("log archive not configured")) + return + } + id := r.PathValue("id") + node := r.PathValue("node") + if id == "" || node == "" { + writeError(w, http.StatusBadRequest, errors.New("execution id and node required")) + return + } + rc, err := s.logs.Get(r.Context(), id, node) + if err != nil { + if errors.Is(err, logstore.ErrNotFound) { + writeError(w, http.StatusNotFound, errors.New("log not found (run may still be in progress)")) + return + } + writeError(w, http.StatusBadGateway, fmt.Errorf("log fetch: %w", err)) + return + } + defer rc.Close() + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.Header().Set("Cache-Control", "no-cache") + if _, err := io.Copy(w, rc); err != nil { + // Connection may have dropped; nothing to do. + return + } +} + // handleListExecutions returns recent runs from storage. Optional // ?pipeline_id= filters by source pipeline; ?limit= caps the page size. func (s *Server) handleListExecutions(w http.ResponseWriter, r *http.Request) { diff --git a/pkg/logstore/logstore.go b/pkg/logstore/logstore.go new file mode 100644 index 0000000..4c5bae4 --- /dev/null +++ b/pkg/logstore/logstore.go @@ -0,0 +1,147 @@ +// Package logstore archives per-node execution logs to S3-compatible +// object storage (MinIO in dev). The API server writes one object per +// node when the node finishes; the live SSE feed continues to stream +// during execution. +// +// Object key layout: +// /runs//.log +package logstore + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "strings" + "time" + + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" +) + +// Store is the interface the API server uses. Implementations include the +// MinIO/S3-backed Mongo (mongo) store and a memory store for tests. +type Store interface { + Put(ctx context.Context, execID, node string, data []byte) error + Get(ctx context.Context, execID, node string) (io.ReadCloser, error) + Exists(ctx context.Context, execID, node string) (bool, error) +} + +// ErrNotFound is returned by Get when the object doesn't exist. +var ErrNotFound = errors.New("log not found") + +// Config carries connection details for a MinIO/S3 endpoint. +type Config struct { + Endpoint string // host:port, no scheme — e.g. "minio:9000" or "127.0.0.1:9000" + AccessKey string + SecretKey string + Bucket string + UseSSL bool +} + +// Minio is the production Store implementation, backed by minio-go. +type Minio struct { + cli *minio.Client + bucket string +} + +// NewMinio dials the endpoint with a 10s timeout and ensures the bucket +// exists. Returns an error if either step fails. +func NewMinio(ctx context.Context, cfg Config) (*Minio, error) { + if cfg.Endpoint == "" { + return nil, errors.New("MINIO_ENDPOINT is required") + } + if cfg.Bucket == "" { + cfg.Bucket = "flow-logs" + } + cli, err := minio.New(cfg.Endpoint, &minio.Options{ + Creds: credentials.NewStaticV4(cfg.AccessKey, cfg.SecretKey, ""), + Secure: cfg.UseSSL, + }) + if err != nil { + return nil, fmt.Errorf("minio client: %w", err) + } + + bucketCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + exists, err := cli.BucketExists(bucketCtx, cfg.Bucket) + if err != nil { + return nil, fmt.Errorf("minio bucket check %q: %w", cfg.Bucket, err) + } + if !exists { + if err := cli.MakeBucket(bucketCtx, cfg.Bucket, minio.MakeBucketOptions{}); err != nil { + return nil, fmt.Errorf("minio make bucket %q: %w", cfg.Bucket, err) + } + } + return &Minio{cli: cli, bucket: cfg.Bucket}, nil +} + +func (m *Minio) Put(ctx context.Context, execID, node string, data []byte) error { + if execID == "" || node == "" { + return errors.New("execID and node required") + } + key := keyFor(execID, node) + _, err := m.cli.PutObject(ctx, m.bucket, key, + bytes.NewReader(data), int64(len(data)), + minio.PutObjectOptions{ContentType: "text/plain; charset=utf-8"}) + return err +} + +func (m *Minio) Get(ctx context.Context, execID, node string) (io.ReadCloser, error) { + key := keyFor(execID, node) + obj, err := m.cli.GetObject(ctx, m.bucket, key, minio.GetObjectOptions{}) + if err != nil { + return nil, err + } + // minio-go returns a lazy *Object; calling Stat surfaces NoSuchKey. + if _, err := obj.Stat(); err != nil { + _ = obj.Close() + var er minio.ErrorResponse + if errors.As(err, &er) && er.Code == "NoSuchKey" { + return nil, ErrNotFound + } + return nil, err + } + return obj, nil +} + +func (m *Minio) Exists(ctx context.Context, execID, node string) (bool, error) { + key := keyFor(execID, node) + _, err := m.cli.StatObject(ctx, m.bucket, key, minio.StatObjectOptions{}) + if err == nil { + return true, nil + } + var er minio.ErrorResponse + if errors.As(err, &er) && er.Code == "NoSuchKey" { + return false, nil + } + return false, err +} + +// keyFor returns the object key for a given execution+node pair. We sanitise +// the node name so weird characters (slashes, spaces) don't bork S3 keys. +func keyFor(execID, node string) string { + return "runs/" + execID + "/" + sanitize(node) + ".log" +} + +func sanitize(s string) string { + // Replace anything that isn't alnum/-/_ with "-". + var b strings.Builder + for _, r := range s { + switch { + case r >= 'a' && r <= 'z', + r >= 'A' && r <= 'Z', + r >= '0' && r <= '9', + r == '-', r == '_', r == '.': + b.WriteRune(r) + default: + b.WriteRune('-') + } + } + out := b.String() + if out == "" { + out = "node" + } + return out +} diff --git a/web/app/agents/view/page.tsx b/web/app/agents/view/page.tsx index 528c3f4..1a4c82d 100644 --- a/web/app/agents/view/page.tsx +++ b/web/app/agents/view/page.tsx @@ -88,6 +88,12 @@ function AgentDetail() { useEffect(() => { load(); + // Reload recent runs whenever any run is dispatched (manual / agent / + // GitHub push). Cheap — `load()` is one round-trip. + const es = new EventSource(api.runsStreamURL()); + es.onmessage = () => load(); + es.onerror = () => {}; + return () => es.close(); // eslint-disable-next-line react-hooks/exhaustive-deps }, [id]); diff --git a/web/app/approvals/page.tsx b/web/app/approvals/page.tsx new file mode 100644 index 0000000..120c6b3 --- /dev/null +++ b/web/app/approvals/page.tsx @@ -0,0 +1,10 @@ +import { EmptySection } from "@/components/empty-section"; + +export default function ApprovalsPage() { + return ( + + ); +} diff --git a/web/app/dashboard/page.tsx b/web/app/dashboard/page.tsx new file mode 100644 index 0000000..ce8231c --- /dev/null +++ b/web/app/dashboard/page.tsx @@ -0,0 +1,10 @@ +import { EmptySection } from "@/components/empty-section"; + +export default function DashboardPage() { + return ( + + ); +} diff --git a/web/app/environments/page.tsx b/web/app/environments/page.tsx new file mode 100644 index 0000000..6c018d1 --- /dev/null +++ b/web/app/environments/page.tsx @@ -0,0 +1,10 @@ +import { EmptySection } from "@/components/empty-section"; + +export default function EnvironmentsPage() { + return ( + + ); +} diff --git a/web/app/executions/view/page.tsx b/web/app/executions/view/page.tsx index a561ff3..ac4ddb4 100644 --- a/web/app/executions/view/page.tsx +++ b/web/app/executions/view/page.tsx @@ -3,7 +3,7 @@ import { Suspense, useEffect, useMemo, useRef, useState } from "react"; import Link from "next/link"; import { useSearchParams } from "next/navigation"; -import { ArrowLeft, RefreshCw, Send } from "lucide-react"; +import { ChevronDown, ChevronRight, RefreshCw, Send } from "lucide-react"; import { Button } from "@/components/ui/button"; import { Card, @@ -18,25 +18,26 @@ import { Label } from "@/components/ui/label"; import { Textarea } from "@/components/ui/textarea"; import { PipelineCanvas } from "@/components/canvas/pipeline-canvas"; import { api, type ExecutionStatus, type Run } from "@/lib/api"; -import type { PipelineDefinition } from "@/lib/pipeline-graph"; +import type { PipelineDefinition, PipelineNode } from "@/lib/pipeline-graph"; +import { lookup as lookupNode } from "@/lib/node-catalog"; +import { formatDate } from "@/lib/utils"; type NodeStatus = "pending" | "running" | "success" | "failed" | "paused"; type NodeStatuses = Record; +type NodeLogs = Record; +type NodeDurations = Record; interface ExecutionEvent { type: string; node?: string; node_type?: string; status?: string; - content?: string; // for node_log events + content?: string; outputs?: Record; error?: string; duration_ms?: number; } -type NodeLogLine = { ts: number; line: string }; -type NodeLogs = Record; - export default function ExecutionPage() { return ( (null); const [pipelineDef, setPipelineDef] = useState(null); const [nodeStatuses, setNodeStatuses] = useState({}); - const [events, setEvents] = useState([]); const [nodeLogs, setNodeLogs] = useState({}); - const [activeLogNode, setActiveLogNode] = useState(null); + const [nodeDurations, setNodeDurations] = useState({}); const [streamConnected, setStreamConnected] = useState(false); const [error, setError] = useState(null); - const [tab, setTab] = useState<"canvas" | "json">("canvas"); // Resume form const [awakeable, setAwakeable] = useState(""); const [data, setData] = useState(`{"approved": true}`); const [resuming, setResuming] = useState(false); - const eventSourceRef = useRef(null); - - // Initial load: status + run record + pipeline def. + // Initial load useEffect(() => { if (!id) return; (async () => { @@ -105,7 +102,7 @@ function ExecutionView() { const f = await api.getFlow(r.pipelineId); setPipelineDef((f.definition as PipelineDefinition | null) ?? null); } catch { - // Pipeline may have been deleted; render JSON view only. + /* pipeline may have been deleted */ } } } catch (e) { @@ -114,41 +111,24 @@ function ExecutionView() { })(); }, [id]); - // SSE subscription. On open, mark all known nodes as pending. Each - // node_started/completed/error event flips that node's status. + // SSE useEffect(() => { if (!id) return; - const url = api.executionStreamURL(id); - const es = new EventSource(url); - eventSourceRef.current = es; + const es = new EventSource(api.executionStreamURL(id)); es.onopen = () => setStreamConnected(true); - es.onerror = () => { - setStreamConnected(false); - // EventSource auto-reconnects on transient errors; only close on - // permanent ones. We log but don't bail. - }; + es.onerror = () => setStreamConnected(false); es.onmessage = (msg) => { try { const ev: ExecutionEvent = JSON.parse(msg.data); - - // Log lines are high-frequency — keep them out of the generic - // events array (used for the JSON debug view) and route into - // their own state instead. if (ev.type === "node_log" && ev.node && typeof ev.content === "string") { const node = ev.node; setNodeLogs((prev) => { const cur = prev[node] ?? []; const next = cur.length >= 2000 ? cur.slice(-1999) : cur; - return { - ...prev, - [node]: [...next, { ts: Date.now(), line: ev.content! }], - }; + return { ...prev, [node]: [...next, ev.content!] }; }); - setActiveLogNode((prev) => prev ?? node); return; } - - setEvents((prev) => [...prev.slice(-99), ev]); if (ev.node) { setNodeStatuses((prev) => { const next: NodeStatus = @@ -161,8 +141,13 @@ function ExecutionView() { : (prev[ev.node!] ?? "pending"); return { ...prev, [ev.node!]: next }; }); - if (ev.type === "node_started") { - setActiveLogNode(ev.node); + if (ev.type === "node_completed" || ev.type === "node_error") { + if (typeof ev.duration_ms === "number") { + setNodeDurations((prev) => ({ + ...prev, + [ev.node!]: ev.duration_ms!, + })); + } } } if (ev.type === "done") { @@ -171,26 +156,44 @@ function ExecutionView() { setStreamConnected(false); } } catch { - // Ignore malformed messages. + /* ignore */ } }; - return () => { es.close(); - eventSourceRef.current = null; setStreamConnected(false); }; }, [id]); - // Fallback polling for terminal status — useful if SSE failed to connect. + // Fallback polling when SSE drops useEffect(() => { if (!id || streamConnected) return; const t = setInterval(() => { api.getExecution(id).then(setStatus).catch(() => {}); - }, 3000); + }, 4000); return () => clearInterval(t); }, [id, streamConnected]); + // Hydrate canvas node statuses from the terminal payload whenever we + // have status + pipeline def. This handles the "joined after the run + // finished" case — without it, the canvas stays at PENDING forever + // because SSE has nothing left to replay. SSE-derived statuses are not + // overwritten so an in-flight run is still authoritative. + useEffect(() => { + if (!status || !pipelineDef) return; + const overall = String(status.status || "").toLowerCase(); + const isTerm = ["success", "completed", "failed", "error", "partial_error"].includes(overall); + if (!isTerm) return; + setNodeStatuses((prev) => { + const next: NodeStatuses = { ...prev }; + for (const n of pipelineDef.nodes ?? []) { + if (next[n.name]) continue; // SSE wins + next[n.name] = inferStatus(status, n.name); + } + return next; + }); + }, [status, pipelineDef]); + async function onResume() { setResuming(true); setError(null); @@ -207,9 +210,14 @@ function ExecutionView() { } } - const overallStatus = useMemo(() => { - return (status?.status as string) || run?.status || "running"; - }, [status, run]); + const overallStatus = useMemo( + () => (status?.status as string) || run?.status || "running", + [status, run] + ); + const isTerminal = ["success", "completed", "failed", "error", "partial_error"] + .includes(overallStatus.toLowerCase()); + + const nodes: PipelineNode[] = pipelineDef?.nodes ?? []; if (!id) { return ( @@ -220,55 +228,147 @@ function ExecutionView() { } return ( -
- {/* Toolbar */} -
- -
- Execution +
+ {/* Header */} +
+
+

Run

+
{id}
+ {run?.pipelineId && ( + + pipeline {run.pipelineId.slice(0, 8)} + + )}
- {id} - {overallStatus} - {streamConnected ? ( - - ● live - - ) : ( - - polling - - )} -
-
- - -
+ /> + {overallStatus} + +
+ + {error && ( + + {error} + + )} + + {/* Pipeline canvas card */} + + + Pipeline + + {nodes.length} {nodes.length === 1 ? "node" : "nodes"} + + + +
+ {pipelineDef ? ( + + ) : ( +
+ {run?.pipelineId + ? "Loading pipeline canvas…" + : "No pipeline definition for this run."} +
+ )} +
+ + + + {/* Nodes panel */} + + + Nodes + + Per-node detail. Build logs stream live and archive to S3 when the + node finishes. + + + + {nodes.length === 0 ? ( +

No nodes loaded.

+ ) : ( + nodes.map((n) => ( + | undefined} + /> + )) + )} + + + + {/* Resume overlay if paused */} + {(overallStatus === "paused" || overallStatus === "waiting") && ( + + + Resume + + Resolve a Restate awakeable to continue. + + + +
+ + setAwakeable(e.target.value)} + placeholder="awk_…" + className="font-mono text-xs" + /> +
+
+ +