diff --git a/README.md b/README.md index 305acdd..19183ff 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # flowctl -flowctl is a **production pipeline orchestrator** for Stellar data components. It manages sources, processors, and sinks through an embedded control plane so you can define pipelines declaratively, run them as supervised processes, and observe their health and status. +flowctl orchestrates Stellar data pipelines with a control plane for component registration, run history, and health. In process mode, it launches components and supervises gRPC delivery between sources, optional processors, and sinks. ## What flowctl is for @@ -9,7 +9,8 @@ flowctl is the **orchestration/production layer** in the ecosystem: - resolve and launch components - run with an embedded or external control plane - track registration, heartbeats, and run state -- inspect status and active runs +- fail a run on delivery errors and cancel its other stream edges +- inspect status, active runs, and recorded failure context If you are rapidly prototyping processors or doing ad-hoc local analysis, use **nebu** first. When a processor is ready to be operated, monitored, and reused in production, it should live behind **flowctl**. @@ -20,7 +21,7 @@ The primary supported runtime path today is: - **Embedded control plane** (default) - **`flowctl/v1` pipelines** -Container/deployment backends exist in the repo but should be treated as secondary until they are hardened and fully tested. +Container/deployment backends exist in the repo but should be treated as secondary until they are hardened and fully tested. Process-mode forwarding currently supports a sequential processor chain (or direct source-to-sink delivery) and fan-out to sinks; it is not an arbitrary DAG runtime. ## Installation @@ -79,7 +80,7 @@ flowctl run stellar-pipeline.yaml In another terminal: ```bash -# 5. Check component health +# 5. Check component health (while the control plane is running) flowctl status # 6. Inspect active runs @@ -104,6 +105,15 @@ flowctl pipelines run-info flowctl pipelines stop ``` +`flowctl run` waits for delivery to complete or for an operator to stop it. A +bounded run completes successfully once every edge closes and sinks acknowledge; +a source, processor, or sink delivery failure marks the run failed and exits +nonzero. An unexpected EOF from an unbounded source is a failure. Currently +bounded completion is inferred from the source's `END_LEDGER` environment +variable; a source with a different finite stopping condition will be treated +as unbounded. On interruption or `pipelines stop`, the run is stopped rather +than failed. Restart and checkpoints remain the operator's responsibility. + ### Recommended operator flags ```bash @@ -209,7 +219,11 @@ Full reference: └──────────────┘ └──────────────┘ └──────────────┘ ``` -**Components are separate programs** that you build or install independently: +**Components are separate programs** that you build or install independently. A +component must register a reachable gRPC endpoint and implement its source, +processor, or consumer stream service to participate in supervised delivery; +registration and heartbeats alone do not make a data path healthy. + - **Sources**: Produce data (e.g., Stellar ledger streams, Kafka consumers) - **Processors**: Transform data (e.g., extract contract events, filter transactions) - **Sinks**: Consume data (e.g., PostgreSQL, webhooks, file storage) @@ -237,16 +251,16 @@ The JavaScript SDK also includes a working `flowctl` process-mode end-to-end exa ## Configuration -Pipelines are configured using Kubernetes-style YAML files. Here's a minimal example: +Pipelines are configured using Kubernetes-style YAML files. This example shows the shape of a process pipeline; the three `command` paths must point to binaries that implement the corresponding gRPC stream services: ```yaml apiVersion: flowctl/v1 kind: Pipeline metadata: name: hello-world-pipeline - description: A simple pipeline demonstrating flowctl basics spec: + description: A simple pipeline demonstrating flowctl basics driver: process # Run components as local processes sources: @@ -259,12 +273,14 @@ spec: processors: - id: data-transformer command: ["./bin/data-transformer"] + inputs: [data-generator] env: TRANSFORM_TYPE: "uppercase" sinks: - id: console-logger command: ["./bin/console-logger"] + inputs: [data-transformer] env: LOG_FORMAT: "pretty" ``` @@ -272,8 +288,33 @@ spec: **Key configuration concepts**: - `apiVersion: flowctl/v1` - Standard pipeline format - `spec.driver` - Execution environment (currently focus on `process`) -- Component `command` - Full path to the component binary +- Component `command` - Executable path (relative paths resolve from the working directory where `flowctl run` is invoked) - Component `env` - Environment variables for configuration +- Component `inputs` - Upstream component IDs; processors are wired in list order in process mode + +Processors are optional: when none are defined, flowctl wires each source +directly to sinks that list its ID in `inputs`. Multiple sinks receive matching event types; a slow +sink applies backpressure rather than causing dropped events. Use +`spec.registrationOnly: true` **only** for intentional monitor-only pipelines: +it disables stream wiring and delivery supervision, so `RUNNING` then means +components registered, not that events are flowing. For example: + +```yaml +spec: + registrationOnly: true # intentional control-plane registration/health only + sources: + - id: my-source + command: ["/path/to/source"] + sinks: + - id: my-sink + inputs: [my-source] + command: ["/path/to/sink"] +``` + +A real source-to-sink pipeline should omit `registrationOnly` and supply +stream-capable components. For now, bounded sources are recognized by +`END_LEDGER` in the source's `env` (including when a component reference +translates `config.end_ledger` into that environment variable). For more examples, see: - `examples/quickstart/testnet-duckdb-pipeline.yaml` - Starter DuckDB pipeline @@ -283,10 +324,9 @@ For more examples, see: ### Logging -flowctl uses structured logging powered by Uber's Zap library. You can control the log level using: - -- Configuration file: Set the `log_level` field in your YAML configuration -- Command line: Use the `--log-level` flag (e.g., `--log-level=debug`) +flowctl uses structured logging powered by Uber's Zap library. Use the +`--log-level` flag (e.g., `--log-level=debug`) to control **flowctl's** logs; +component binaries may have separate logging settings. Available log levels: - `debug`: Detailed information for debugging @@ -317,18 +357,17 @@ If you specifically need Docker Compose translation, local execution generation, ### DAG-Based Processing -flowctl supports a Directed Acyclic Graph (DAG) based processing pipeline, which allows complex topologies beyond simple linear chains. This enables: - -- Parallel processing of data streams -- Fan-out/fan-in patterns -- Buffered channels for flow control -- Strongly typed event connections - -For more information, see [DAG-Based Processing](docs/dag-processing.md). +The repository also contains DAG execution code, but the primary `flowctl run` +process path currently wires processors in list order and fans out to sinks. +It does not implement arbitrary DAG fan-in/fan-out or buffered sink queues. +See [Architecture: DAG event flow](docs/architecture.md#event-flow-dag---diamond-pattern) for the separate DAG model. ### Secure Communication with TLS -flowctl supports secure communication between components using Transport Layer Security (TLS): +The repository has TLS configuration for server/client integrations. **The primary +process-mode stream orchestrator currently connects to component gRPC endpoints +using insecure transport.** Do not assume the TLS settings below protect the +`flowctl run` source-to-sink data path. The available TLS options include: - Server-side TLS for encrypted communication - Mutual TLS (mTLS) where both client and server authenticate each other @@ -344,18 +383,8 @@ flowctl server --port 8080 --tls-cert server.crt --tls-key server.key flowctl server --port 8080 --tls-cert server.crt --tls-key server.key --tls-ca-cert ca.crt ``` -For client configuration, add a `tls` section to your YAML configuration: -```yaml -tls: - mode: "enabled" # Options: disabled, enabled, mutual - cert_file: "client.crt" - key_file: "client.key" - ca_file: "ca.crt" # Required for mutual TLS - skip_verify: false - server_name: "flowctl-server.example.com" # For SNI verification -``` - -For more information, see [TLS Configuration](docs/tls-configuration.md). +These server flags do not enable TLS on the process-mode component streams. +For background, see [Architecture: Communication Security](docs/architecture.md#communication-security). ## Processor Discovery @@ -415,9 +444,9 @@ Discover and inspect processors registered with the control plane at runtime. ls -la /path/to/component chmod +x /path/to/component -# 2. Verify absolute path is used (required for driver: process) -# ✅ Good: /home/user/bin/my-component -# ❌ Bad: ./bin/my-component +# 2. Confirm the command path is valid from flowctl's working directory +# Absolute paths are less sensitive to where you start flowctl. +# Relative paths such as ./bin/my-component also work from that directory. # 3. Test component standalone /path/to/component @@ -440,15 +469,18 @@ chmod +x /path/to/component ./bin/flowctl run --control-plane-port 9090 pipeline.yaml ./bin/flowctl run --control-plane-port 0 pipeline.yaml -# 3. Verify ENABLE_FLOWCTL is set -# In pipeline YAML: +# 3. Check the component's registration, heartbeat, and service logs. +# SDK packages: github.com/withObsrvr/flowctl-sdk/pkg/{source,processor,consumer} +``` + +A component must register a reachable gRPC stream endpoint. Some SDK-based +components require these pipeline environment settings (check that component's +configuration and use the actual control-plane address): + +```yaml env: - ENABLE_FLOWCTL: "true" + ENABLE_FLOWCTL: "true" # Component-specific, not universal FLOWCTL_ENDPOINT: "127.0.0.1:8080" - -# 4. Check component built with flowctl-sdk -# Components must use SDK for registration: -# github.com/withObsrvr/flowctl-sdk/pkg/{source,processor,consumer} ``` #### No Data Flowing Between Components @@ -458,20 +490,14 @@ env: **Solutions:** ```bash -# 1. Verify inputs are correctly specified -processors: - - id: my-processor - inputs: ["correct-source-id"] # Must match source id +1. Verify that `inputs` names the upstream component ID. Without processors, + a sink can name a source directly. Check registered event types too: sinks + with a nonempty input type list receive only matching events. +2. Check `flowctl run`'s exit code and the failing edge in its output. A failed + edge cancels the other streams rather than dropping events: -# 2. Check event types match -# Source OutputType must match Processor InputType -# Check component code or logs - -# 3. Enable debug logging +```bash ./bin/flowctl run pipeline.yaml --log-level=debug - -# 4. Check for processing errors -./bin/flowctl run pipeline.yaml 2>&1 | grep -i error ``` #### Port Already in Use @@ -524,8 +550,7 @@ kill # - apiVersion: flowctl/v1 # - kind: Pipeline # - metadata.name -# - spec.driver -# - At least one source and one sink +# - At least one source and one sink (also required in registration-only mode) # 3. Verify YAML syntax # Use yamllint or online YAML validator @@ -568,14 +593,14 @@ CONFIG_KEY=value \ # Active runs on a specific control plane ./bin/flowctl pipelines active --control-plane-address 127.0.0.1 --control-plane-port 9090 -# Inspect one run in detail (full id or unique prefix) +# Inspect one run in detail (full id or unique prefix, including delivery error) ./bin/flowctl pipelines run-info abc12345 --control-plane-address 127.0.0.1 --control-plane-port 9090 # Stop a running pipeline ./bin/flowctl pipelines stop abc12345 --control-plane-address 127.0.0.1 --control-plane-port 9090 ``` -**Persistence note:** embedded control plane mode now uses BoltDB-backed storage by default, so run history can survive process restarts. Use `--db-path` to choose the storage location or `--no-persistence` for ephemeral local runs. +**Persistence note:** embedded mode uses BoltDB at `~/.flowctl/flowctl-service-registry.db` by default. Use `--db-path` to choose a file or `--no-persistence` for an ephemeral run. A failed run stores the first delivery error as a structured, credential-free record in the run's `Error` field; `run-info` displays it. `pipelines run-info` requires a *running* control plane: after an embedded run exits, start a standalone `flowctl server --db-path ` (with no other process using the DB) to inspect persisted history. ### Docker and Container Issues @@ -634,7 +659,7 @@ If you're still having trouble: ### Sandbox Issues -For sandbox-specific issues, refer to [docs/sandbox.md](docs/sandbox.md) for detailed troubleshooting steps. +For sandbox and container-related issues, see [Docker and translation workflows](docs/docker-and-translation.md). ## Development @@ -673,6 +698,7 @@ make run-example ### Advanced Topics - **[Architecture](docs/architecture.md)** - System design, data flow, and component lifecycle +- **[Delivery modernization](docs/modernization-plan.md)** - Supervised edges, error reporting, and current limitations - **[Performance Tuning](docs/performance-tuning.md)** - Optimization strategies and scaling ### Real-World Example diff --git a/docs/modernization-plan.md b/docs/modernization-plan.md new file mode 100644 index 0000000..2eca7fd --- /dev/null +++ b/docs/modernization-plan.md @@ -0,0 +1,153 @@ +# flowctl modernization plan + +**Role:** Pipeline supervisor and stream orchestrator +**Local checkout:** `/home/tillman/Documents/flowctl` +**Depends on:** Tagged error-aware `flowctl-sdk` release and a source that uses it + +## Implementation notes (current checkout) + +The runner now waits for supervised source/processor/sink edges and returns +transport failures as CLI errors. Bounded completion is inferred from the +source's `END_LEDGER` environment setting; an unbounded source EOF is an error. +The first failing edge cancels all streams. Fan-out sends synchronously, so a +slow sink applies backpressure instead of dropping events. The run is marked +failed on delivery error, completed on bounded completion, and stopped on +operator cancellation. Restart and checkpoints remain owned by the operator. + +The process-driver synthetic source test in `test/delivery_process_test.go` +exercises nonzero CLI exit, failure before and after one event, EOF, bounded +completion, processor/sink failures, and operator cancellation. The live Stellar +process smoke completed successfully with 101 events at the sink; it requires +prebuilt flowctl and SDK processor binaries and network access. + +## Direct wiring and durable failure records + +Pipelines without processors now wire `source → sink` only for sinks that +list the source ID in `inputs`, and supervise delivery with the same +cancellation and completion semantics. Components that only register with +the control plane must opt out explicitly using +`spec.registrationOnly: true`. In this mode, `RUNNING` means registered, **not** +that events are being delivered; use it only for intentional monitor-only +pipelines. The operator smoke test uses this explicit mode. + +On delivery failure the runner writes a structured, credential-free JSON +record (component, from/to, edge, operation, classified cause) to the existing +`PipelineRun.error` field. Both storage backends persist this field; +`flowctl pipelines run-info` already displays it. The first error record is +kept if further status updates occur. Component heartbeat health metadata +remains a live signal, not the authoritative historical failure reason. + +## Remaining limitation + +**Boundedness is inferred from `END_LEDGER`.** Sources with other finite +termination criteria are currently classified as unbounded and their normal +EOF fails the run. Introduce an explicit bounded-source setting before +supporting other finite source types. + +## Objective + +Make pipeline status reflect delivery reality. A failed source, processor, +transport edge, or sink must fail and cancel the run instead of being logged by +a detached goroutine while the pipeline appears healthy. + +## Historical failure boundaries (before this implementation) + +Previously `internal/runner/stream_orchestrator.go` launched forwarding +goroutines and `WireAll` returned after wiring, not after delivery completed. +Observed behavior included: + +- source `Recv` errors are logged and discarded; +- processor `Recv` and `Send` errors are logged and discarded; +- sink `Send` errors may be counted and processing continues; +- sink backpressure timeout drops an event and processing continues; +- `CloseAndRecv` errors are ignored; +- goroutine termination is not delivered to `PipelineRunner`; +- a pipeline can remain apparently alive after its only data path exits. + +## Required outcome + +Introduce one orchestrator-owned terminal result mechanism. An +`errgroup.WithContext` or equivalent first-error-wins design should: + +- attach every forwarding loop to the pipeline context; +- return component and edge identity with every failure; +- cancel all other loops on the first non-cancellation failure; +- wait for all loops to exit; +- return the terminal result to `PipelineRunner`; +- set run/component state to failed; +- cause `flowctl run` to exit nonzero; +- preserve successful bounded completion; +- treat unexpected EOF from an unbounded source as failure. + +No event may be dropped to keep the pipeline alive. Backpressure must slow the +pipeline or fail it explicitly. + +## Pull-request sequence + +1. Add characterization tests for source, processor, and sink edge failures. +2. Add an orchestrator terminal-error channel/group and expose `Wait`. +3. Move source-to-processor forwarding under that lifecycle. +4. Move processor chaining and sink fan-out under the same lifecycle. +5. Remove log-and-continue delivery behavior. +6. Connect terminal result to `PipelineRunner`, run status, and CLI exit code. +7. Add process-driver E2E with a synthetic source that emits once then fails. +8. Document bounded completion, unbounded cancellation, and restart ownership. + +## Error context + +Every returned/logged delivery error should carry: + +- pipeline/run ID; +- source and destination component IDs; +- edge type; +- operation (`recv`, `send`, `close`, registration); +- last event ID/type when safe; +- classified cause. + +Never log credentials or complete environment maps. + +## Required tests + +- source fails before first event; +- source emits one event then fails; +- bounded source completes successfully; +- unbounded source ends unexpectedly; +- processor receive failure; +- processor send failure; +- one fan-out sink fails while another is healthy; +- sink buffer remains full; +- sink `CloseAndRecv` fails; +- operator cancellation returns interrupted rather than failed; +- all forwarding goroutines exit; +- CLI returns nonzero for pipeline failure. + +## Verification + +```bash +cd /home/tillman/Documents/flowctl +go test ./... +go vet ./... +go build ./... +``` + +Then run the process-driver smoke from: + +```bash +cd /home/tillman/Documents/stellar-raw-ledger-origin +RUN_TIMEOUT=45s scripts/flowctl-process-e2e.sh +``` + +## Non-goals + +- component-specific retry loops; +- source checkpoints; +- automatic infinite restart; +- changing Stellar event schemas; +- container-orchestrator redesign; +- dropping events as a congestion strategy. + +## Definition of done + +A source terminal error crosses gRPC, reaches `PipelineRunner`, cancels the +pipeline, marks the run failed, stops all forwarding goroutines, and makes +`flowctl run` exit nonzero with the failing edge identified. diff --git a/internal/api/control_plane.go b/internal/api/control_plane.go index 64eede9..af8090b 100644 --- a/internal/api/control_plane.go +++ b/internal/api/control_plane.go @@ -271,7 +271,7 @@ func (s *ControlPlaneServer) Heartbeat(ctx context.Context, req *flowctlv1.Heart now := time.Now() service.LastSeen = now service.Status.Metrics = mergeStringMaps(service.Status.Metrics, req.Metrics) - if req.Status != flowctlv1.HealthStatus_HEALTH_STATUS_UNKNOWN { + if req.Status != flowctlv1.HealthStatus_HEALTH_STATUS_UNKNOWN && !(service.Status.Metrics["delivery_run_id"] != "" && req.Status == flowctlv1.HealthStatus_HEALTH_STATUS_HEALTHY) { service.Status.Status = req.Status } else if service.Status.Status == flowctlv1.HealthStatus_HEALTH_STATUS_UNKNOWN { service.Status.Status = flowctlv1.HealthStatus_HEALTH_STATUS_HEALTHY @@ -283,7 +283,7 @@ func (s *ControlPlaneServer) Heartbeat(ctx context.Context, req *flowctlv1.Heart err := s.storage.UpdateService(ctx, req.ServiceId, func(storedService *storage.ServiceInfo) error { storedService.LastSeen = now storedService.Status.Metrics = mergeStringMaps(storedService.Status.Metrics, req.Metrics) - if req.Status != flowctlv1.HealthStatus_HEALTH_STATUS_UNKNOWN { + if req.Status != flowctlv1.HealthStatus_HEALTH_STATUS_UNKNOWN && !(storedService.Status.Metrics["delivery_run_id"] != "" && req.Status == flowctlv1.HealthStatus_HEALTH_STATUS_HEALTHY) { storedService.Status.Status = req.Status } else if storedService.Status.Status == flowctlv1.HealthStatus_HEALTH_STATUS_UNKNOWN { storedService.Status.Status = flowctlv1.HealthStatus_HEALTH_STATUS_HEALTHY @@ -569,7 +569,7 @@ func (s *ControlPlaneServer) UpdatePipelineRun(ctx context.Context, req *flowctl if req.Metrics != nil { runInfo.Run.Metrics = req.Metrics } - if req.Error != "" { + if req.Error != "" && runInfo.Run.Error == "" { runInfo.Run.Error = req.Error } updatedRun = runInfo.Run diff --git a/internal/api/delivery_health_test.go b/internal/api/delivery_health_test.go new file mode 100644 index 0000000..407c906 --- /dev/null +++ b/internal/api/delivery_health_test.go @@ -0,0 +1,39 @@ +package api + +import ( + "context" + "testing" + + flowctlv1 "github.com/withObsrvr/flow-proto/go/gen/flowctl/v1" + "github.com/withobsrvr/flowctl/internal/storage" +) + +func TestDeliveryFailureHealthSurvivesHeartbeat(t *testing.T) { + store := storage.NewMemoryStorage() + if err := store.Open(); err != nil { + t.Fatal(err) + } + defer store.Close() + server := NewControlPlaneServer(store) + ctx := context.Background() + _, err := server.RegisterComponent(ctx, &flowctlv1.RegisterRequest{ComponentId: "source", Component: &flowctlv1.ComponentInfo{Id: "source"}}) + if err != nil { + t.Fatal(err) + } + _, err = server.Heartbeat(ctx, &flowctlv1.HeartbeatRequest{ServiceId: "source", Status: flowctlv1.HealthStatus_HEALTH_STATUS_UNHEALTHY, Metrics: map[string]string{"delivery_run_id": "run-1"}}) + if err != nil { + t.Fatal(err) + } + _, err = server.Heartbeat(ctx, &flowctlv1.HeartbeatRequest{ServiceId: "source", Status: flowctlv1.HealthStatus_HEALTH_STATUS_HEALTHY}) + if err != nil { + t.Fatal(err) + } + resp, err := server.GetComponentStatus(ctx, &flowctlv1.ComponentStatusRequest{ServiceId: "source"}) + if err != nil || resp.Status != flowctlv1.HealthStatus_HEALTH_STATUS_UNHEALTHY { + t.Fatalf("status=%v err=%v", resp, err) + } + saved, err := store.GetService(ctx, "source") + if err != nil || saved.Status.Status != flowctlv1.HealthStatus_HEALTH_STATUS_UNHEALTHY { + t.Fatalf("stored status=%v err=%v", saved, err) + } +} diff --git a/internal/api/delivery_run_error_test.go b/internal/api/delivery_run_error_test.go new file mode 100644 index 0000000..3ac84ee --- /dev/null +++ b/internal/api/delivery_run_error_test.go @@ -0,0 +1,34 @@ +package api + +import ( + "context" + "testing" + + "github.com/withobsrvr/flowctl/internal/storage" + flowctlpb "github.com/withobsrvr/flowctl/proto" +) + +func TestFirstRunErrorPersists(t *testing.T) { + store := storage.NewMemoryStorage() + if err := store.Open(); err != nil { + t.Fatal(err) + } + defer store.Close() + server := NewControlPlaneServer(store) + ctx := context.Background() + _, err := server.CreatePipelineRun(ctx, &flowctlpb.CreatePipelineRunRequest{RunId: "run-1", PipelineName: "delivery"}) + if err != nil { + t.Fatal(err) + } + first := `{"component":"source","edge":"source-sink","cause":"UNEXPECTED_EOF"}` + for _, value := range []string{first, `{"component":"sink"}`} { + _, err = server.UpdatePipelineRun(ctx, &flowctlpb.UpdatePipelineRunRequest{RunId: "run-1", Status: flowctlpb.RunStatus_RUN_STATUS_FAILED, Error: value}) + if err != nil { + t.Fatal(err) + } + } + run, err := server.GetPipelineRun(ctx, &flowctlpb.GetPipelineRunRequest{RunId: "run-1"}) + if err != nil || run.Error != first { + t.Fatalf("run error=%q err=%v", run.GetError(), err) + } +} diff --git a/internal/model/pipeline.go b/internal/model/pipeline.go index 57390ba..1fc0728 100644 --- a/internal/model/pipeline.go +++ b/internal/model/pipeline.go @@ -1,6 +1,5 @@ package model - // Pipeline represents a data processing pipeline with all components type Pipeline struct { APIVersion string `yaml:"apiVersion" json:"apiVersion"` @@ -19,36 +18,37 @@ type Metadata struct { // Spec contains the pipeline specification type Spec struct { - Description string `yaml:"description,omitempty" json:"description,omitempty"` - Driver string `yaml:"driver,omitempty" json:"driver,omitempty"` - Sources []Component `yaml:"sources" json:"sources"` - Processors []Component `yaml:"processors" json:"processors"` - Sinks []Component `yaml:"sinks" json:"sinks"` - Pipelines []Component `yaml:"pipelines,omitempty" json:"pipelines,omitempty"` // New: complete pipelines as components - Config map[string]interface{} `yaml:"config,omitempty" json:"config,omitempty"` + Description string `yaml:"description,omitempty" json:"description,omitempty"` + Driver string `yaml:"driver,omitempty" json:"driver,omitempty"` + RegistrationOnly bool `yaml:"registrationOnly,omitempty" json:"registrationOnly,omitempty"` + Sources []Component `yaml:"sources" json:"sources"` + Processors []Component `yaml:"processors" json:"processors"` + Sinks []Component `yaml:"sinks" json:"sinks"` + Pipelines []Component `yaml:"pipelines,omitempty" json:"pipelines,omitempty"` // New: complete pipelines as components + Config map[string]interface{} `yaml:"config,omitempty" json:"config,omitempty"` } // Component represents a pipeline component (source, processor, sink, or pipeline) type Component struct { - ID string `yaml:"id" json:"id"` - Type string `yaml:"type,omitempty" json:"type,omitempty"` - Image string `yaml:"image,omitempty" json:"image,omitempty"` // OCI image reference (alternative to Command) - Command []string `yaml:"command,omitempty" json:"command,omitempty"` // Local binary path (alternative to Image) - Args []string `yaml:"args,omitempty" json:"args,omitempty"` // Arguments for image/command - Inputs []string `yaml:"inputs,omitempty" json:"inputs,omitempty"` - InputEventTypes []string `yaml:"input_event_types,omitempty" json:"input_event_types,omitempty"` - OutputEventTypes []string `yaml:"output_event_types,omitempty" json:"output_event_types,omitempty"` - Config map[string]interface{} `yaml:"config,omitempty" json:"config,omitempty"` - Env map[string]string `yaml:"env,omitempty" json:"env,omitempty"` - Volumes []Volume `yaml:"volumes,omitempty" json:"volumes,omitempty"` - Secrets []SecretMount `yaml:"secrets,omitempty" json:"secrets,omitempty"` // New: secret mounts - Ports []Port `yaml:"ports,omitempty" json:"ports,omitempty"` - HealthCheck string `yaml:"health_check,omitempty" json:"health_check,omitempty"` - HealthPort int `yaml:"health_port,omitempty" json:"health_port,omitempty"` - Replicas int `yaml:"replicas,omitempty" json:"replicas,omitempty"` - Resources Resources `yaml:"resources,omitempty" json:"resources,omitempty"` - DependsOn []string `yaml:"depends_on,omitempty" json:"depends_on,omitempty"` // New: pipeline dependencies - RestartPolicy string `yaml:"restart_policy,omitempty" json:"restart_policy,omitempty"` // New: restart policy for pipelines + ID string `yaml:"id" json:"id"` + Type string `yaml:"type,omitempty" json:"type,omitempty"` + Image string `yaml:"image,omitempty" json:"image,omitempty"` // OCI image reference (alternative to Command) + Command []string `yaml:"command,omitempty" json:"command,omitempty"` // Local binary path (alternative to Image) + Args []string `yaml:"args,omitempty" json:"args,omitempty"` // Arguments for image/command + Inputs []string `yaml:"inputs,omitempty" json:"inputs,omitempty"` + InputEventTypes []string `yaml:"input_event_types,omitempty" json:"input_event_types,omitempty"` + OutputEventTypes []string `yaml:"output_event_types,omitempty" json:"output_event_types,omitempty"` + Config map[string]interface{} `yaml:"config,omitempty" json:"config,omitempty"` + Env map[string]string `yaml:"env,omitempty" json:"env,omitempty"` + Volumes []Volume `yaml:"volumes,omitempty" json:"volumes,omitempty"` + Secrets []SecretMount `yaml:"secrets,omitempty" json:"secrets,omitempty"` // New: secret mounts + Ports []Port `yaml:"ports,omitempty" json:"ports,omitempty"` + HealthCheck string `yaml:"health_check,omitempty" json:"health_check,omitempty"` + HealthPort int `yaml:"health_port,omitempty" json:"health_port,omitempty"` + Replicas int `yaml:"replicas,omitempty" json:"replicas,omitempty"` + Resources Resources `yaml:"resources,omitempty" json:"resources,omitempty"` + DependsOn []string `yaml:"depends_on,omitempty" json:"depends_on,omitempty"` // New: pipeline dependencies + RestartPolicy string `yaml:"restart_policy,omitempty" json:"restart_policy,omitempty"` // New: restart policy for pipelines } // Volume represents a volume mount in a component @@ -90,4 +90,4 @@ type SecretMount struct { Type string `yaml:"type" json:"type"` Required bool `yaml:"required" json:"required"` Labels map[string]string `yaml:"labels" json:"labels"` -} \ No newline at end of file +} diff --git a/internal/runner/delivery_shutdown_test.go b/internal/runner/delivery_shutdown_test.go new file mode 100644 index 0000000..3181edf --- /dev/null +++ b/internal/runner/delivery_shutdown_test.go @@ -0,0 +1,34 @@ +package runner + +import ( + "errors" + "testing" +) + +func TestWaitForTerminalOnStop(t *testing.T) { + failure := errors.New("edge failed") + t.Run("failure already reported", func(t *testing.T) { + terminal := make(chan error, 1) + terminal <- failure + cancelled := false + err, reaped := waitForTerminalOnStop(terminal, func() { cancelled = true }) + if !reaped || !errors.Is(err, failure) || cancelled { + t.Fatalf("err=%v reaped=%t cancelled=%t", err, reaped, cancelled) + } + }) + t.Run("failure captured during cancellation", func(t *testing.T) { + terminal := make(chan error, 1) + cancelled := false + err, reaped := waitForTerminalOnStop(terminal, func() { cancelled = true; terminal <- failure }) + if !reaped || !errors.Is(err, failure) || !cancelled { + t.Fatalf("err=%v reaped=%t cancelled=%t", err, reaped, cancelled) + } + }) + t.Run("operator cancellation", func(t *testing.T) { + terminal := make(chan error, 1) + err, reaped := waitForTerminalOnStop(terminal, func() { terminal <- nil }) + if !reaped || err != nil { + t.Fatalf("err=%v reaped=%t", err, reaped) + } + }) +} diff --git a/internal/runner/pipeline_runner.go b/internal/runner/pipeline_runner.go index 7bb33a1..3cde007 100644 --- a/internal/runner/pipeline_runner.go +++ b/internal/runner/pipeline_runner.go @@ -2,7 +2,10 @@ package runner import ( "context" + "encoding/json" + "errors" "fmt" + "io" "net" "os" "os/signal" @@ -23,7 +26,9 @@ import ( flowctlpb "github.com/withobsrvr/flowctl/proto" "go.uber.org/zap" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/emptypb" ) @@ -210,9 +215,7 @@ func (r *PipelineRunner) Run() error { fmt.Println("→ Launching components...") if err := r.startComponents(); err != nil { r.updateRunStatus(flowctlpb.RunStatus_RUN_STATUS_FAILED) - if r.controlPlane != nil { - r.controlPlane.Stop() - } + _ = r.shutdown() return err } @@ -220,9 +223,7 @@ func (r *PipelineRunner) Run() error { fmt.Println("→ Waiting for components to register...") if err := r.waitForComponentRegistration(); err != nil { r.updateRunStatus(flowctlpb.RunStatus_RUN_STATUS_FAILED) - if r.controlPlane != nil { - r.controlPlane.Stop() - } + _ = r.shutdown() return err } @@ -232,12 +233,31 @@ func (r *PipelineRunner) Run() error { fmt.Println("→ Wiring data flow between components...") streamOrch := NewStreamOrchestrator(r.ctx, r.controlPlane, r.pipeline) - if err := streamOrch.WireAll(); err != nil { - r.updateRunStatus(flowctlpb.RunStatus_RUN_STATUS_FAILED) - if r.controlPlane != nil { - r.controlPlane.Stop() + streamOrch.runID = r.runID + streamOrch.controlPlaneEndpoint = r.GetControlPlaneEndpoint() + // Registration-only mode is an explicit opt-out of delivery supervision. + registrationOnly := r.pipeline.Spec.RegistrationOnly + if !registrationOnly { + if err := streamOrch.WireAll(); err != nil { + groupErr := streamOrch.Stop() + // A previously started edge can fail while later sources are still + // wiring. Preserve that delivery error rather than a secondary cancel. + if groupErr != nil && (errors.Is(err, context.Canceled) || status.Code(err) == codes.Canceled) { + err = groupErr + } + r.markDeliveryFailure(err) + r.updateRunStatusWithError(flowctlpb.RunStatus_RUN_STATUS_FAILED, err) + _ = r.shutdown() + return fmt.Errorf("failed to wire components: %w", err) } - return fmt.Errorf("failed to wire components: %w", err) + } + + defer streamOrch.Stop() + terminal := make(chan error, 1) + if !registrationOnly { + go func() { terminal <- streamOrch.Wait() }() + } else { + logger.Warn("Registration-only pipeline: delivery is not supervised") } // Update status to RUNNING now that pipeline is fully wired @@ -273,9 +293,19 @@ func (r *PipelineRunner) Run() error { sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) - // Wait for shutdown + defer signal.Stop(sigChan) + // Wait for shutdown or for all delivery edges to terminate. var shutdownStatus flowctlpb.RunStatus + var deliveryErr error select { + case deliveryErr = <-terminal: + if deliveryErr != nil { + shutdownStatus = flowctlpb.RunStatus_RUN_STATUS_FAILED + } else if r.ctx.Err() != nil { + shutdownStatus = flowctlpb.RunStatus_RUN_STATUS_STOPPED + } else { + shutdownStatus = flowctlpb.RunStatus_RUN_STATUS_COMPLETED + } case <-sigChan: logger.Info("Received shutdown signal") shutdownStatus = flowctlpb.RunStatus_RUN_STATUS_STOPPED @@ -284,14 +314,82 @@ func (r *PipelineRunner) Run() error { if r.stopRequested.Load() { shutdownStatus = flowctlpb.RunStatus_RUN_STATUS_STOPPED } else { - shutdownStatus = flowctlpb.RunStatus_RUN_STATUS_COMPLETED + shutdownStatus = flowctlpb.RunStatus_RUN_STATUS_STOPPED } } + // Reap delivery before stopping components. Prefer an already-reported + // failure over a simultaneous interrupt, but don't fail on cancellation. + if shutdownStatus == flowctlpb.RunStatus_RUN_STATUS_STOPPED && !registrationOnly { + var reaped bool + deliveryErr, reaped = waitForTerminalOnStop(terminal, streamOrch.cancel) + if !reaped { + logger.Warn("Timed out waiting for forwarding loops to exit") + } + if deliveryErr != nil { + shutdownStatus = flowctlpb.RunStatus_RUN_STATUS_FAILED + } + } + if deliveryErr != nil { + logger.Error("Pipeline delivery failed", zap.Error(deliveryErr), zap.String("run_id", r.runID)) + r.markDeliveryFailure(deliveryErr) + } // Update run status before shutdown - r.updateRunStatus(shutdownStatus) + r.updateRunStatusWithError(shutdownStatus, deliveryErr) - return r.shutdown() + shutdownErr := r.shutdown() + if deliveryErr != nil { + return deliveryErr + } + return shutdownErr +} + +// waitForTerminalOnStop preserves a failure already reported or captured by +// the errgroup while cancellation is in progress. A cancelled edge returns nil. +func waitForTerminalOnStop(terminal <-chan error, cancel func()) (error, bool) { + select { + case err := <-terminal: + return err, true + default: + } + cancel() + select { + case err := <-terminal: + return err, true + case <-time.After(10 * time.Second): + return nil, false + } +} + +// markDeliveryFailure records which registered component failed. Store only +// structured edge metadata and the gRPC cause code, never payloads or env vars. +func (r *PipelineRunner) markDeliveryFailure(err error) { + var edge *DeliveryError + if !errors.As(err, &edge) { + return + } + id := edge.From + if edge.Operation == "send" || edge.Operation == "close" || edge.Operation == "connect" || edge.Operation == "registration" || edge.Operation == "open" { + id = edge.To + } + if id == "sinks" { + id = edge.From + } + client, conn, connectErr := r.getV1ControlPlaneClient() + if connectErr != nil { + logger.Warn("Cannot mark failed component", zap.Error(connectErr)) + return + } + defer conn.Close() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _, heartbeatErr := client.Heartbeat(ctx, &flowctlv1.HeartbeatRequest{ + ServiceId: id, Status: flowctlv1.HealthStatus_HEALTH_STATUS_UNHEALTHY, + Metrics: map[string]string{"delivery_run_id": r.runID, "delivery_edge": edge.Kind, "delivery_operation": edge.Operation, "delivery_cause": status.Code(edge.Err).String()}, + }) + if heartbeatErr != nil { + logger.Warn("Cannot mark failed component", zap.String("component", id), zap.Error(heartbeatErr)) + } } // startComponents starts all pipeline components in dependency order @@ -950,8 +1048,37 @@ func (r *PipelineRunner) Stop() { }) } -// updateRunStatus updates the run status in the control plane +// updateRunStatus updates the run status in the control plane. func (r *PipelineRunner) updateRunStatus(status flowctlpb.RunStatus) { + r.updateRunStatusWithError(status, nil) +} + +// The existing PipelineRun.error protobuf field is persisted in both storage +// backends. Keep this record structured and free of raw server messages, which +// may contain secrets; the full error remains available on the CLI at exit. +func deliveryFailureRecord(err error) string { + if err == nil { + return "" + } + var edge *DeliveryError + if !errors.As(err, &edge) { + return "" + } + cause := status.Code(edge.Err).String() + if errors.Is(edge.Err, io.ErrUnexpectedEOF) { + cause = "UNEXPECTED_EOF" + } else if errors.Is(edge.Err, io.EOF) { + cause = "EOF" + } + fields := map[string]string{"component": edge.From, "from": edge.From, "to": edge.To, "edge": edge.Kind, "operation": edge.Operation, "cause": cause} + if edge.Operation == "send" || edge.Operation == "close" || edge.Operation == "open" || edge.Operation == "connect" || edge.Operation == "registration" { + fields["component"] = edge.To + } + encoded, _ := json.Marshal(fields) + return string(encoded) +} + +func (r *PipelineRunner) updateRunStatusWithError(status flowctlpb.RunStatus, failure error) { if r.runID == "" { // No run ID means run tracking failed to initialize return @@ -989,6 +1116,7 @@ func (r *PipelineRunner) updateRunStatus(status flowctlpb.RunStatus) { RunId: r.runID, Status: status, Metrics: metrics, + Error: deliveryFailureRecord(failure), } _, err = client.UpdatePipelineRun(r.ctx, req) diff --git a/internal/runner/stream_orchestrator.go b/internal/runner/stream_orchestrator.go index 9124c03..5907191 100644 --- a/internal/runner/stream_orchestrator.go +++ b/internal/runner/stream_orchestrator.go @@ -2,737 +2,380 @@ package runner import ( "context" + "errors" "fmt" "io" - "os" - "strconv" "sync" "time" - "go.uber.org/zap" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - flowctlv1 "github.com/withObsrvr/flow-proto/go/gen/flowctl/v1" "github.com/withobsrvr/flowctl/internal/controlplane" "github.com/withobsrvr/flowctl/internal/model" - "github.com/withobsrvr/flowctl/internal/utils/logger" + "golang.org/x/sync/errgroup" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" ) -// getSinkBufferSize returns the buffer size for sink channels from environment or default -func getSinkBufferSize() int { - if envVal := os.Getenv("FLOWCTL_SINK_BUFFER_SIZE"); envVal != "" { - if size, err := strconv.Atoi(envVal); err == nil && size > 0 { - return size - } +// DeliveryError identifies the edge and operation that failed without exposing +// component configuration or credentials. +type DeliveryError struct { + Pipeline, RunID, From, To, Kind, Operation string + EventID, EventType string + Err error +} + +func (e *DeliveryError) Error() string { + context := fmt.Sprintf("pipeline %s/%s edge %s %s -> %s %s", e.Pipeline, e.RunID, e.Kind, e.From, e.To, e.Operation) + if e.EventID != "" { + context += fmt.Sprintf(" event %s (%s)", e.EventID, e.EventType) } - return 1000 // Default buffer size + return context + ": " + e.Err.Error() } +func (e *DeliveryError) Unwrap() error { return e.Err } -// StreamOrchestrator manages data flow between pipeline components +// StreamOrchestrator owns every transport edge until delivery completes or fails. type StreamOrchestrator struct { - controlPlane *controlplane.EmbeddedControlPlane - pipeline *model.Pipeline - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup - connections map[string]*grpc.ClientConn - processorStreams map[string]flowctlv1.ProcessorService_ProcessClient - mu sync.Mutex + controlPlane *controlplane.EmbeddedControlPlane + controlPlaneEndpoint string + pipeline *model.Pipeline + runID string + ctx context.Context + cancel context.CancelFunc + group *errgroup.Group + connections map[string]*grpc.ClientConn + mu sync.Mutex + started bool } -// NewStreamOrchestrator creates a new stream orchestrator -func NewStreamOrchestrator(ctx context.Context, controlPlane *controlplane.EmbeddedControlPlane, pipeline *model.Pipeline) *StreamOrchestrator { - orchCtx, cancel := context.WithCancel(ctx) - return &StreamOrchestrator{ - controlPlane: controlPlane, - pipeline: pipeline, - ctx: orchCtx, - cancel: cancel, - connections: make(map[string]*grpc.ClientConn), - processorStreams: make(map[string]flowctlv1.ProcessorService_ProcessClient), - } +func NewStreamOrchestrator(ctx context.Context, cp *controlplane.EmbeddedControlPlane, pipeline *model.Pipeline) *StreamOrchestrator { + ctx, cancel := context.WithCancel(ctx) + group, groupCtx := errgroup.WithContext(ctx) + return &StreamOrchestrator{controlPlane: cp, pipeline: pipeline, ctx: groupCtx, cancel: cancel, group: group, connections: make(map[string]*grpc.ClientConn)} } -// WireAll wires all components in the pipeline together -func (s *StreamOrchestrator) WireAll() error { - logger.Info("Starting stream orchestrator to wire components") - - // Get all registered components from control plane - services, err := s.controlPlane.GetServiceList() - if err != nil { - return fmt.Errorf("failed to get service list: %w", err) +func (s *StreamOrchestrator) edge(from, to, kind, operation string, event *flowctlv1.Event, err error) error { + e := &DeliveryError{Pipeline: s.pipeline.Metadata.Name, RunID: s.runID, From: from, To: to, Kind: kind, Operation: operation, Err: err} + if event != nil { + e.EventID, e.EventType = event.Id, event.Type } + return e +} - // Build a map of component ID to registered info - componentMap := make(map[string]*flowctlv1.ComponentInfo) - for _, service := range services { - if service.Component != nil { - componentMap[service.Component.Id] = service.Component +func (s *StreamOrchestrator) launch(fn func() error) { + s.group.Go(func() error { + err := fn() + if err != nil && s.ctx.Err() != nil && (errors.Is(err, context.Canceled) || status.Code(err) == codes.Canceled || status.Code(err) == codes.DeadlineExceeded) { + return nil } - } + return err + }) +} - // Wire sources to first processors (or all if no chaining) +// WireAll opens the streams and starts supervised forwarding. Wait must be called +// after WireAll, including when wiring fails, to reap already-started edges. +func (s *StreamOrchestrator) WireAll() error { + if s.started { + return fmt.Errorf("orchestrator already wired") + } + s.started = true + if len(s.pipeline.Spec.Sources) == 0 || len(s.pipeline.Spec.Sinks) == 0 { + return fmt.Errorf("pipeline %s requires sources and sinks for stream wiring (or explicit registrationOnly mode)", s.pipeline.Metadata.Name) + } for _, source := range s.pipeline.Spec.Sources { - _, ok := componentMap[source.ID] - if !ok { - logger.Error("Source not found in registered components", zap.String("source", source.ID)) - continue - } - - // Find first processor (or processors that accept this source's output) - var targetProcessors []model.Component + target := "sinks" if len(s.pipeline.Spec.Processors) > 0 { - // CYCLE 2: Use first processor in chain - targetProcessors = []model.Component{s.pipeline.Spec.Processors[0]} - logger.Info("Wiring source to first processor in chain", - zap.String("source", source.ID), - zap.String("first_processor", targetProcessors[0].ID)) + target = s.pipeline.Spec.Processors[0].ID } - - for _, processor := range targetProcessors { - logger.Info("Wiring source to processor", - zap.String("source", source.ID), - zap.String("processor", processor.ID)) - - s.wg.Add(1) - go func(src model.Component, proc model.Component) { - defer s.wg.Done() - processStream, err := s.wireSourceToProcessor(src, proc) - if err != nil { - logger.Error("Failed to wire source to processor", - zap.Error(err), - zap.String("source", src.ID), - zap.String("processor", proc.ID)) - return - } - - // Store the processor stream for downstream components - s.mu.Lock() - s.processorStreams[proc.ID] = processStream - s.mu.Unlock() - }(source, processor) + srcInfo, err := s.getComponentEndpoint(source.ID) + if err != nil { + s.cancel() + return s.edge(source.ID, target, "source-processor", "registration", nil, err) + } + srcConn, err := s.getConnection(srcInfo.Endpoint) + if err != nil { + s.cancel() + return s.edge(source.ID, target, "source-processor", "connect", nil, err) + } + params := map[string]string{} + if v := source.Env["START_LEDGER"]; v != "" { + params["start_ledger"] = v + } + if v := source.Env["END_LEDGER"]; v != "" { + params["end_ledger"] = v + } + sourceStream, err := flowctlv1.NewSourceServiceClient(srcConn).StreamEvents(s.ctx, &flowctlv1.StreamRequest{Params: params}) + if err != nil { + s.cancel() + return s.edge(source.ID, target, "source-processor", "open", nil, err) } - } - - // CYCLE 2: Wire processor chains (proc[i] -> proc[i+1]) - for i := 0; i < len(s.pipeline.Spec.Processors)-1; i++ { - currentProc := s.pipeline.Spec.Processors[i] - nextProc := s.pipeline.Spec.Processors[i+1] - - logger.Info("Wiring processor chain", - zap.String("from", currentProc.ID), - zap.String("to", nextProc.ID), - zap.Int("chain_position", i)) - - s.wg.Add(1) - go func(curr, next model.Component) { - defer s.wg.Done() - - // Wait for current processor stream to be available - var currentStream flowctlv1.ProcessorService_ProcessClient - for { - s.mu.Lock() - ps, ok := s.processorStreams[curr.ID] - s.mu.Unlock() - - if ok { - currentStream = ps - break - } - // Wait for upstream wiring to complete - select { - case <-s.ctx.Done(): - return - case <-time.After(50 * time.Millisecond): - // Continue loop - } + // Each source owns its own processor streams. Sharing a bidi stream between + // sources would allow concurrent Send/Recv and obscure which edge failed. + streams := make([]flowctlv1.ProcessorService_ProcessClient, len(s.pipeline.Spec.Processors)) + for i, proc := range s.pipeline.Spec.Processors { + info, err := s.getComponentEndpoint(proc.ID) + if err != nil { + s.cancel() + return s.edge(source.ID, proc.ID, "processor", "registration", nil, err) } - - // Wire current processor to next processor - nextStream, err := s.wireProcessorToProcessor(curr, next, currentStream) + conn, err := s.getConnection(info.Endpoint) if err != nil { - logger.Error("Failed to wire processor chain", - zap.Error(err), - zap.String("from", curr.ID), - zap.String("to", next.ID)) - return + s.cancel() + return s.edge(source.ID, proc.ID, "processor", "connect", nil, err) } - - // Store next processor stream for downstream - s.mu.Lock() - s.processorStreams[next.ID] = nextStream - s.mu.Unlock() - }(currentProc, nextProc) - } - - // Wire last processor to sinks (FAN-OUT) - if len(s.pipeline.Spec.Processors) > 0 && len(s.pipeline.Spec.Sinks) > 0 { - lastProcessor := s.pipeline.Spec.Processors[len(s.pipeline.Spec.Processors)-1] - - logger.Info("Wiring last processor to sinks (fan-out)", - zap.String("processor", lastProcessor.ID), - zap.Int("num_sinks", len(s.pipeline.Spec.Sinks))) - - s.wg.Add(1) - go func(proc model.Component, sinks []model.Component) { - defer s.wg.Done() - - // Wait for last processor stream to be available - var processStream flowctlv1.ProcessorService_ProcessClient - for { - s.mu.Lock() - ps, ok := s.processorStreams[proc.ID] - s.mu.Unlock() - - if ok { - processStream = ps - break - } - - // Wait for processor chain to complete - select { - case <-s.ctx.Done(): - return - case <-time.After(50 * time.Millisecond): - // Continue loop - } + streams[i], err = flowctlv1.NewProcessorServiceClient(conn).Process(s.ctx) + if err != nil { + s.cancel() + return s.edge(source.ID, proc.ID, "processor", "open", nil, err) } - - // CYCLE 2: Fan out to ALL sinks in parallel (broadcast pattern) - if err := s.fanOutToSinks(proc, sinks, processStream); err != nil { - logger.Error("Failed to fan out to sinks", - zap.Error(err), - zap.String("processor", proc.ID)) + } + last := source.ID + kind := "source-sink" + if len(streams) > 0 { + last = s.pipeline.Spec.Processors[len(streams)-1].ID + kind = "processor-sink" + } + sinks := make([]sinkStream, 0, len(s.pipeline.Spec.Sinks)) + for _, sink := range s.pipeline.Spec.Sinks { + // Without processors, inputs describe the only valid source-to-sink + // edges. Event-type filtering is applied separately after routing. + if len(streams) == 0 && !hasInput(sink.Inputs, source.ID) { + continue } - }(lastProcessor, s.pipeline.Spec.Sinks) - } - - logger.Info("Stream orchestrator wired all components") - return nil -} - -// wireSourceToProcessor connects a source to a processor -// Returns the processor stream so it can be shared with downstream components -func (s *StreamOrchestrator) wireSourceToProcessor(source, processor model.Component) (flowctlv1.ProcessorService_ProcessClient, error) { - // Get source endpoint from control plane - sourceInfo, err := s.getComponentEndpoint(source.ID) - if err != nil { - return nil, fmt.Errorf("failed to get source endpoint: %w", err) - } - - // Get processor endpoint from control plane - processorInfo, err := s.getComponentEndpoint(processor.ID) - if err != nil { - return nil, fmt.Errorf("failed to get processor endpoint: %w", err) - } - - // Connect to source - sourceConn, err := s.getConnection(sourceInfo.Endpoint) - if err != nil { - return nil, fmt.Errorf("failed to connect to source: %w", err) - } - - // Connect to processor - processorConn, err := s.getConnection(processorInfo.Endpoint) - if err != nil { - return nil, fmt.Errorf("failed to connect to processor: %w", err) - } - - // Create clients - sourceClient := flowctlv1.NewSourceServiceClient(sourceConn) - processorClient := flowctlv1.NewProcessorServiceClient(processorConn) - - // Set large message limits for all streams (50MB) - maxMsgSize := 50 * 1024 * 1024 // 50MB - - logger.Info("Creating source stream with large message limits", - zap.Int("maxMsgSize", maxMsgSize), - zap.String("source", source.ID)) - - // Build params from source component's environment variables - params := make(map[string]string) - - // Pass START_LEDGER and END_LEDGER as stream parameters - if startLedger, ok := source.Env["START_LEDGER"]; ok { - params["start_ledger"] = startLedger - logger.Info("Passing start_ledger to source stream", zap.String("start_ledger", startLedger)) - } - if endLedger, ok := source.Env["END_LEDGER"]; ok { - params["end_ledger"] = endLedger - logger.Info("Passing end_ledger to source stream", zap.String("end_ledger", endLedger)) - } - - // Start streaming from source (limits set on connection) - stream, err := sourceClient.StreamEvents(s.ctx, &flowctlv1.StreamRequest{ - Params: params, - Resume: false, - }) - if err != nil { - return nil, fmt.Errorf("failed to start source stream: %w", err) - } - - logger.Info("Creating processor stream with large message limits", - zap.Int("maxMsgSize", maxMsgSize), - zap.String("processor", processor.ID)) - - // Create bidirectional processor stream (limits set on connection) - processStream, err := processorClient.Process(s.ctx) - if err != nil { - return nil, fmt.Errorf("failed to start processor stream: %w", err) - } - - logger.Info("Started streaming from source to processor", - zap.String("source", source.ID), - zap.String("processor", processor.ID)) - - // Read from source and write to processor in one goroutine - go func() { - for { - envelope, err := stream.Recv() - if err == io.EOF { - logger.Info("Source stream ended", zap.String("source", source.ID)) - processStream.CloseSend() - return + info, err := s.getComponentEndpoint(sink.ID) + if err != nil { + s.cancel() + return s.edge(last, sink.ID, kind, "registration", nil, err) } + conn, err := s.getConnection(info.Endpoint) if err != nil { - logger.Error("Error receiving from source", zap.Error(err), zap.String("source", source.ID)) - processStream.CloseSend() - return + s.cancel() + return s.edge(last, sink.ID, kind, "connect", nil, err) } - - // Send event to processor - if err := processStream.Send(envelope); err != nil { - logger.Error("Error sending to processor", zap.Error(err), zap.String("processor", processor.ID)) - processStream.CloseSend() - return + stream, err := flowctlv1.NewConsumerServiceClient(conn).Consume(s.ctx) + if err != nil { + s.cancel() + return s.edge(last, sink.ID, kind, "open", nil, err) } + sinks = append(sinks, sinkStream{id: sink.ID, accepted: info.InputEventTypes, stream: stream}) } - }() - - // Return the processor stream so it can be read by downstream components - return processStream, nil -} - -// wireProcessorToProcessor connects one processor to another processor (for chaining) -// Returns the next processor's stream for further chaining -func (s *StreamOrchestrator) wireProcessorToProcessor(currentProc, nextProc model.Component, currentStream flowctlv1.ProcessorService_ProcessClient) (flowctlv1.ProcessorService_ProcessClient, error) { - // Get next processor endpoint from control plane - nextProcInfo, err := s.getComponentEndpoint(nextProc.ID) - if err != nil { - return nil, fmt.Errorf("failed to get next processor endpoint: %w", err) - } - - // Connect to next processor - nextProcConn, err := s.getConnection(nextProcInfo.Endpoint) - if err != nil { - return nil, fmt.Errorf("failed to connect to next processor: %w", err) + if len(streams) == 0 { + if len(sinks) == 0 { + s.cancel() + return fmt.Errorf("pipeline %s source %s has no sink declaring it in inputs", s.pipeline.Metadata.Name, source.ID) + } + bounded := params["end_ledger"] != "" + s.launch(func() error { + return s.fanOut(sourceReceiver{sourceStream, s.ctx, bounded}, source.ID, sinks, "source-sink") + }) + continue + } + first := s.pipeline.Spec.Processors[0].ID + bounded := params["end_ledger"] != "" + s.launch(func() error { return s.forwardSource(sourceStream, streams[0], source.ID, first, bounded) }) + for i := 0; i < len(streams)-1; i++ { + from, to := s.pipeline.Spec.Processors[i].ID, s.pipeline.Spec.Processors[i+1].ID + current, next := streams[i], streams[i+1] + s.launch(func() error { return s.forward(current, next, from, to) }) + } + s.launch(func() error { return s.fanOut(streams[len(streams)-1], last, sinks) }) } + return nil +} - // Create next processor client - nextProcClient := flowctlv1.NewProcessorServiceClient(nextProcConn) - - // Create bidirectional stream for next processor - nextProcessStream, err := nextProcClient.Process(s.ctx) - if err != nil { - return nil, fmt.Errorf("failed to start next processor stream: %w", err) +func hasInput(inputs []string, id string) bool { + for _, input := range inputs { + if input == id { + return true + } } + return false +} - logger.Info("Started processor chain link", - zap.String("from", currentProc.ID), - zap.String("to", nextProc.ID)) - - // Forward events from current processor to next processor - go func() { - for { - envelope, err := currentStream.Recv() - if err == io.EOF { - logger.Info("Processor stream ended in chain", zap.String("processor", currentProc.ID)) - nextProcessStream.CloseSend() - return - } - if err != nil { - logger.Error("Error receiving from processor in chain", zap.Error(err), zap.String("processor", currentProc.ID)) - nextProcessStream.CloseSend() - return +func (s *StreamOrchestrator) forwardSource(from interface { + Recv() (*flowctlv1.Event, error) +}, to flowctlv1.ProcessorService_ProcessClient, fromID, toID string, bounded bool) error { + for { + event, err := from.Recv() + if err == io.EOF { + if !bounded && s.ctx.Err() == nil { + return s.edge(fromID, toID, "source-processor", "recv (unexpected EOF)", nil, io.ErrUnexpectedEOF) } - - // Send event to next processor - if err := nextProcessStream.Send(envelope); err != nil { - logger.Error("Error sending to next processor in chain", zap.Error(err), zap.String("next_processor", nextProc.ID)) - nextProcessStream.CloseSend() - return + if err := to.CloseSend(); err != nil { + return s.edge(fromID, toID, "source-processor", "close", nil, err) } + return nil } - }() - - // Return the next processor stream for further chaining or sink wiring - return nextProcessStream, nil -} - -// fanOutToSinks broadcasts events from a processor to multiple sinks in parallel -// CYCLE 2: Supports event type-based routing -// Uses dedicated writer goroutines per sink to ensure thread-safe gRPC stream access -func (s *StreamOrchestrator) fanOutToSinks(processor model.Component, sinks []model.Component, processStream flowctlv1.ProcessorService_ProcessClient) error { - // Create sink connections and streams - type sinkConnection struct { - sink model.Component - stream flowctlv1.ConsumerService_ConsumeClient - acceptedEventTypes []string // CYCLE 2: Event type filtering - eventChan chan *flowctlv1.Event // Channel for serializing sends - } - - var sinkConns []sinkConnection - - // Connect to all sinks first - for _, sink := range sinks { - sinkInfo, err := s.getComponentEndpoint(sink.ID) if err != nil { - logger.Error("Failed to get sink endpoint for fan-out", zap.Error(err), zap.String("sink", sink.ID)) - continue + return s.edge(fromID, toID, "source-processor", "recv", nil, err) } - - sinkConn, err := s.getConnection(sinkInfo.Endpoint) - if err != nil { - logger.Error("Failed to connect to sink for fan-out", zap.Error(err), zap.String("sink", sink.ID)) - continue + if err := to.Send(event); err != nil { + return s.edge(fromID, toID, "source-processor", "send", event, err) } + } +} - sinkClient := flowctlv1.NewConsumerServiceClient(sinkConn) - consumeStream, err := sinkClient.Consume(s.ctx) +func (s *StreamOrchestrator) forward(from, to flowctlv1.ProcessorService_ProcessClient, fromID, toID string) error { + for { + event, err := from.Recv() + if err == io.EOF { + if err := to.CloseSend(); err != nil { + return s.edge(fromID, toID, "processor-processor", "close", nil, err) + } + return nil + } if err != nil { - logger.Error("Failed to start consumer stream for fan-out", zap.Error(err), zap.String("sink", sink.ID)) - continue + return s.edge(fromID, toID, "processor-processor", "recv", nil, err) } - - // CYCLE 2: Get accepted event types from sink registration - acceptedTypes := sinkInfo.InputEventTypes - if len(acceptedTypes) > 0 { - logger.Info("Sink accepts specific event types", - zap.String("sink", sink.ID), - zap.Strings("event_types", acceptedTypes)) - } else { - logger.Info("Sink accepts all event types", - zap.String("sink", sink.ID)) + if err := to.Send(event); err != nil { + return s.edge(fromID, toID, "processor-processor", "send", event, err) } - - bufferSize := getSinkBufferSize() - sinkConns = append(sinkConns, sinkConnection{ - sink: sink, - stream: consumeStream, - acceptedEventTypes: acceptedTypes, - eventChan: make(chan *flowctlv1.Event, bufferSize), - }) - logger.Info("Connected sink for fan-out", - zap.String("sink", sink.ID), - zap.Int("buffer_size", bufferSize)) } +} - if len(sinkConns) == 0 { - return fmt.Errorf("no sinks successfully connected for fan-out") +type sourceReceiver struct { + stream interface { + Recv() (*flowctlv1.Event, error) } + ctx context.Context + bounded bool +} - logger.Info("Fan-out ready", - zap.String("processor", processor.ID), - zap.Int("num_sinks", len(sinkConns))) - - // WaitGroup to track writer goroutines - var writerWg sync.WaitGroup - - // Start a dedicated writer goroutine for each sink - // This ensures thread-safe access to each gRPC stream (one goroutine per stream) - for i := range sinkConns { - sc := &sinkConns[i] - writerWg.Add(1) - go func(sinkConn *sinkConnection) { - defer writerWg.Done() - var eventsSent int64 - var sendErrors int64 - - for event := range sinkConn.eventChan { - if err := sinkConn.stream.Send(event); err != nil { - sendErrors++ - logger.Error("Error sending to sink in fan-out", - zap.Error(err), - zap.String("sink", sinkConn.sink.ID), - zap.String("event_id", event.Id), - zap.String("event_type", event.Type)) - // Continue processing remaining events - } else { - eventsSent++ - } - } - - // Channel closed, close the stream - logger.Info("Sink writer finished, closing stream", - zap.String("sink", sinkConn.sink.ID), - zap.Int64("events_sent", eventsSent), - zap.Int64("send_errors", sendErrors)) - sinkConn.stream.CloseAndRecv() - }(sc) +func (r sourceReceiver) Recv() (*flowctlv1.Event, error) { + event, err := r.stream.Recv() + if err == io.EOF && r.ctx.Err() != nil { + return nil, r.ctx.Err() } + if err == io.EOF && !r.bounded { + return nil, io.ErrUnexpectedEOF + } + return event, err +} - // Read from processor and broadcast to sink channels - go func() { - eventsFiltered := make(map[string]int64) // Track events filtered per sink - droppedEvents := make(map[string]int64) // Track events dropped per sink due to buffer full - - for { - envelope, err := processStream.Recv() - if err == io.EOF { - logger.Info("Processor stream ended, closing sink channels", zap.String("processor", processor.ID)) - // Log filtering and drop statistics - for _, sc := range sinkConns { - if filtered := eventsFiltered[sc.sink.ID]; filtered > 0 { - logger.Info("Sink filtering statistics", - zap.String("sink", sc.sink.ID), - zap.Int64("events_filtered", filtered)) - } - if dropped := droppedEvents[sc.sink.ID]; dropped > 0 { - logger.Error("Sink dropped events due to buffer pressure", - zap.String("sink", sc.sink.ID), - zap.Int64("events_dropped", dropped)) - } - } - // Close all sink channels to signal writers to finish - for i := range sinkConns { - close(sinkConns[i].eventChan) - } - // Wait for all writers to finish before returning - writerWg.Wait() - return - } - if err != nil { - logger.Error("Error receiving from processor for fan-out", zap.Error(err), zap.String("processor", processor.ID)) - // Close all sink channels to signal writers to finish - for i := range sinkConns { - close(sinkConns[i].eventChan) - } - // Wait for all writers to finish before returning - writerWg.Wait() - return - } - - // CYCLE 2: Broadcast to matching sinks based on event type - for i := range sinkConns { - sc := &sinkConns[i] - // Check if sink accepts this event type - if !s.sinkAcceptsEventType(sc.acceptedEventTypes, envelope.Type) { - eventsFiltered[sc.sink.ID]++ - continue - } - - // Send to sink's channel with timeout for backpressure - // This allows the reader to slow down when sinks can't keep up, - // while preventing indefinite blocking - select { - case sc.eventChan <- envelope: - // Event queued successfully - case <-time.After(5 * time.Second): - // Buffer full for extended period - log error and continue - // This preserves backpressure while preventing indefinite blocking - logger.Error("Sink channel full after timeout, event dropped", - zap.String("sink", sc.sink.ID), - zap.String("event_id", envelope.Id), - zap.String("event_type", envelope.Type)) - droppedEvents[sc.sink.ID]++ - } - } - } - }() - - return nil +type sinkStream struct { + id string + accepted []string + stream flowctlv1.ConsumerService_ConsumeClient } -// sinkAcceptsEventType checks if a sink accepts a given event type -// If acceptedTypes is empty, sink accepts all event types -func (s *StreamOrchestrator) sinkAcceptsEventType(acceptedTypes []string, eventType string) bool { - // Empty list means accept all types - if len(acceptedTypes) == 0 { +func accepts(types []string, eventType string) bool { + if len(types) == 0 { return true } - - // Check if event type is in accepted list - for _, acceptedType := range acceptedTypes { - if acceptedType == eventType { + for _, t := range types { + if t == eventType { return true } } - return false } -// wireProcessorToSink connects a processor to a sink using an existing processor stream -func (s *StreamOrchestrator) wireProcessorToSink(processor, sink model.Component, processStream flowctlv1.ProcessorService_ProcessClient) error { - // Get sink endpoint from control plane - sinkInfo, err := s.getComponentEndpoint(sink.ID) - if err != nil { - return fmt.Errorf("failed to get sink endpoint: %w", err) - } - - // Connect to sink - sinkConn, err := s.getConnection(sinkInfo.Endpoint) - if err != nil { - return fmt.Errorf("failed to connect to sink: %w", err) - } - - // Create sink client - sinkClient := flowctlv1.NewConsumerServiceClient(sinkConn) - - // Start consumer stream (limits set on connection) - consumeStream, err := sinkClient.Consume(s.ctx) - if err != nil { - return fmt.Errorf("failed to start consumer stream: %w", err) - } - - logger.Info("Started streaming from processor to sink", - zap.String("processor", processor.ID), - zap.String("sink", sink.ID)) - - // Forward events from processor to sink - go func() { - for { - envelope, err := processStream.Recv() - if err == io.EOF { - logger.Info("Processor stream ended", zap.String("processor", processor.ID)) - consumeStream.CloseAndRecv() - return - } - if err != nil { - logger.Error("Error receiving from processor", zap.Error(err), zap.String("processor", processor.ID)) - consumeStream.CloseAndRecv() - return +// Sending synchronously provides backpressure without dropping events. A blocked +// send is interrupted by the pipeline context when another edge fails. +func (s *StreamOrchestrator) fanOut(from interface { + Recv() (*flowctlv1.Event, error) +}, fromID string, sinks []sinkStream, edgeKind ...string) error { + kind := "processor-sink" + if len(edgeKind) > 0 { + kind = edgeKind[0] + } + for { + event, err := from.Recv() + if err == io.EOF { + for _, sink := range sinks { + response, err := sink.stream.CloseAndRecv() + if err != nil { + return s.edge(fromID, sink.id, kind, "close", nil, err) + } + if response.GetEventsFailed() > 0 || len(response.GetErrors()) > 0 { + return s.edge(fromID, sink.id, kind, "close", nil, fmt.Errorf("sink reported %d failed events (%d errors)", response.GetEventsFailed(), len(response.GetErrors()))) + } } - - // Send event to sink - if err := consumeStream.Send(envelope); err != nil { - logger.Error("Error sending to sink", zap.Error(err), zap.String("sink", sink.ID)) - continue + return nil + } + if err != nil { + return s.edge(fromID, "sinks", kind, "recv", nil, err) + } + for _, sink := range sinks { + if accepts(sink.accepted, event.Type) { + if err := sink.stream.Send(event); err != nil { + // gRPC may report EOF on Send while the actual server error is + // available only from CloseAndRecv. + if errors.Is(err, io.EOF) { + if _, closeErr := sink.stream.CloseAndRecv(); closeErr != nil { + err = closeErr + } + } + return s.edge(fromID, sink.id, kind, "send", event, err) + } } } - }() - - return nil -} - -// getComponentEndpoint gets the endpoint of a component from the control plane -func (s *StreamOrchestrator) getComponentEndpoint(componentID string) (*flowctlv1.ComponentInfo, error) { - // Query control plane for component list - services, err := s.controlPlane.GetServiceList() - if err != nil { - return nil, err } +} - // Find the component by ID - for _, service := range services { - if service.Component != nil && service.Component.Id == componentID { - return service.Component, nil +func (s *StreamOrchestrator) getComponentEndpoint(id string) (*flowctlv1.ComponentInfo, error) { + if s.controlPlane != nil { + services, err := s.controlPlane.GetServiceList() + if err != nil { + return nil, err + } + for _, service := range services { + if service.Component != nil && service.Component.Id == id { + return service.Component, nil + } + } + } else { + conn, err := s.getConnection(s.controlPlaneEndpoint) + if err != nil { + return nil, err + } + ctx, cancel := context.WithTimeout(s.ctx, 5*time.Second) + defer cancel() + resp, err := flowctlv1.NewControlPlaneServiceClient(conn).ListComponents(ctx, &flowctlv1.ListComponentsRequest{IncludeUnhealthy: true}) + if err != nil { + return nil, err + } + for _, service := range resp.Components { + if service.Component != nil && service.Component.Id == id { + return service.Component, nil + } } } - - return nil, fmt.Errorf("component %s not found", componentID) + return nil, fmt.Errorf("component %s not registered", id) } -// getConnection gets or creates a gRPC connection to an endpoint func (s *StreamOrchestrator) getConnection(endpoint string) (*grpc.ClientConn, error) { s.mu.Lock() defer s.mu.Unlock() - - // Check if we already have a connection - if conn, ok := s.connections[endpoint]; ok { - logger.Info("Reusing existing connection", - zap.String("endpoint", endpoint)) + if conn := s.connections[endpoint]; conn != nil { return conn, nil } - - // Create new connection with larger message limits (50MB) - maxMsgSize := 50 * 1024 * 1024 // 50MB - - logger.Info("Creating NEW connection with message limits", - zap.String("endpoint", endpoint), - zap.Int("maxMsgSize", maxMsgSize)) - - conn, err := grpc.NewClient(endpoint, - grpc.WithTransportCredentials(insecure.NewCredentials()), - grpc.WithDefaultCallOptions( - grpc.MaxCallRecvMsgSize(maxMsgSize), - grpc.MaxCallSendMsgSize(maxMsgSize), - ), - ) - if err != nil { - return nil, err - } - - s.connections[endpoint] = conn - logger.Info("Connection created and cached", - zap.String("endpoint", endpoint)) - return conn, nil -} - -// findDownstreamProcessorsFromRegistry finds processors that accept the given event types from control plane registry -func (s *StreamOrchestrator) findDownstreamProcessorsFromRegistry(outputTypes []string, componentMap map[string]*flowctlv1.ComponentInfo) []model.Component { - var processors []model.Component - - for _, processor := range s.pipeline.Spec.Processors { - processorInfo, ok := componentMap[processor.ID] - if !ok { - continue - } - - // Check if processor accepts any of the output types - for _, outputType := range outputTypes { - for _, inputType := range processorInfo.InputEventTypes { - if outputType == inputType { - processors = append(processors, processor) - break - } - } - } + const maxMessage = 50 * 1024 * 1024 + conn, err := grpc.NewClient(endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMessage), grpc.MaxCallSendMsgSize(maxMessage))) + if err == nil { + s.connections[endpoint] = conn } - - return processors + return conn, err } -// findDownstreamSinksFromRegistry finds sinks that accept the given event types from control plane registry -func (s *StreamOrchestrator) findDownstreamSinksFromRegistry(outputTypes []string, componentMap map[string]*flowctlv1.ComponentInfo) []model.Component { - var sinks []model.Component - - for _, sink := range s.pipeline.Spec.Sinks { - sinkInfo, ok := componentMap[sink.ID] - if !ok { - continue - } - - // Check if sink accepts any of the output types - for _, outputType := range outputTypes { - for _, inputType := range sinkInfo.InputEventTypes { - if outputType == inputType { - sinks = append(sinks, sink) - break - } - } - } +// Wait reaps all forwarding loops; cancellation is not a delivery failure. +func (s *StreamOrchestrator) Wait() error { + err := s.group.Wait() + if s.ctx.Err() != nil && (errors.Is(err, context.Canceled) || status.Code(err) == codes.Canceled || err == nil) { + return nil } - - return sinks + return err } -// Stop stops the stream orchestrator and closes all connections func (s *StreamOrchestrator) Stop() error { - logger.Info("Stopping stream orchestrator") - s.cancel() - s.wg.Wait() - - // Close all connections + err := s.Wait() s.mu.Lock() defer s.mu.Unlock() - - for endpoint, conn := range s.connections { - logger.Info("Closing connection", zap.String("endpoint", endpoint)) - conn.Close() + for _, conn := range s.connections { + _ = conn.Close() } - - return nil + return err } diff --git a/internal/runner/stream_orchestrator_test.go b/internal/runner/stream_orchestrator_test.go new file mode 100644 index 0000000..4755aef --- /dev/null +++ b/internal/runner/stream_orchestrator_test.go @@ -0,0 +1,176 @@ +package runner + +import ( + "context" + "errors" + "io" + "strings" + "testing" + + flowctlv1 "github.com/withObsrvr/flow-proto/go/gen/flowctl/v1" + "github.com/withobsrvr/flowctl/internal/model" + "google.golang.org/grpc" +) + +type fakeProcessor struct { + grpc.ClientStream + events []*flowctlv1.Event + recvErr, sendErr, closeErr error + sent int +} + +func (f *fakeProcessor) Recv() (*flowctlv1.Event, error) { + if len(f.events) > 0 { + e := f.events[0] + f.events = f.events[1:] + return e, nil + } + if f.recvErr != nil { + return nil, f.recvErr + } + return nil, io.EOF +} +func (f *fakeProcessor) Send(*flowctlv1.Event) error { f.sent++; return f.sendErr } +func (f *fakeProcessor) CloseSend() error { return f.closeErr } + +type fakeSink struct { + grpc.ClientStream + sent int + sendErr, closeErr error + response *flowctlv1.ConsumeResponse + blocked, release chan struct{} +} + +func (f *fakeSink) Send(*flowctlv1.Event) error { + if f.blocked != nil { + close(f.blocked) + <-f.release + } + f.sent++ + return f.sendErr +} +func (f *fakeSink) CloseAndRecv() (*flowctlv1.ConsumeResponse, error) { return f.response, f.closeErr } + +func testOrchestrator(t *testing.T) *StreamOrchestrator { + t.Helper() + s := NewStreamOrchestrator(context.Background(), nil, &model.Pipeline{}) + t.Cleanup(func() { _ = s.Stop() }) + return s +} +func TestSourceTermination(t *testing.T) { + failure := errors.New("source failed") + for _, tc := range []struct { + name string + events int + recvErr error + bounded bool + want error + }{ + {"before first", 0, failure, false, failure}, + {"after event", 1, failure, false, failure}, + {"bounded completion", 1, nil, true, nil}, + {"unbounded EOF", 1, nil, false, io.ErrUnexpectedEOF}, + } { + t.Run(tc.name, func(t *testing.T) { + s := testOrchestrator(t) + from := &fakeProcessor{recvErr: tc.recvErr} + if tc.events > 0 { + from.events = []*flowctlv1.Event{{Id: "1", Type: "ledger"}} + } + to := &fakeProcessor{} + err := s.forwardSource(from, to, "source", "processor", tc.bounded) + if !errors.Is(err, tc.want) || (tc.want == nil && err != nil) || to.sent != tc.events { + t.Fatalf("err=%v sent=%d", err, to.sent) + } + }) + } +} + +func TestForwardingFailures(t *testing.T) { + failure := errors.New("synthetic failure") + for _, tc := range []struct { + name string + upstream, downstream fakeProcessor + operation string + }{ + {"receive", fakeProcessor{recvErr: failure}, fakeProcessor{}, "recv"}, + {"send", fakeProcessor{events: []*flowctlv1.Event{{Id: "1", Type: "ledger"}}}, fakeProcessor{sendErr: failure}, "send event 1 (ledger)"}, + {"close", fakeProcessor{}, fakeProcessor{closeErr: failure}, "close"}, + } { + t.Run(tc.name, func(t *testing.T) { + s := testOrchestrator(t) + err := s.forward(&tc.upstream, &tc.downstream, "processor-a", "processor-b") + if !errors.Is(err, failure) || !strings.Contains(err.Error(), "processor-a -> processor-b") || !strings.Contains(err.Error(), tc.operation) { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} +func TestFanOutFailuresAndCompletion(t *testing.T) { + failure := errors.New("sink failure") + for _, tc := range []struct { + name string + recvErr, sendErr, closeErr error + want string + }{ + {name: "processor receive", recvErr: failure, want: "recv"}, + {name: "sink send", sendErr: failure, want: "send event 1"}, + {name: "sink close", closeErr: failure, want: "close"}, + {name: "success"}, + {name: "sink reports failed event", want: "reported 1 failed events"}, + } { + t.Run(tc.name, func(t *testing.T) { + s := testOrchestrator(t) + source := &fakeProcessor{events: []*flowctlv1.Event{{Id: "1", Type: "ledger"}}, recvErr: tc.recvErr} + healthy := &fakeSink{} + sink := &fakeSink{sendErr: tc.sendErr, closeErr: tc.closeErr} + if tc.name == "sink reports failed event" { + sink.response = &flowctlv1.ConsumeResponse{EventsFailed: 1} + } + err := s.fanOut(source, "processor", []sinkStream{{id: "healthy", stream: healthy}, {id: "failing", stream: sink}}) + if tc.want == "" { + if err != nil || healthy.sent != 1 || sink.sent != 1 { + t.Fatalf("completion: err=%v healthy=%d sink=%d", err, healthy.sent, sink.sent) + } + return + } + if err == nil || !strings.Contains(err.Error(), tc.want) || (tc.name != "sink reports failed event" && !errors.Is(err, failure)) { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} +func TestFanOutBackpressureDoesNotDropEvent(t *testing.T) { + s := testOrchestrator(t) + slow := &fakeSink{blocked: make(chan struct{}), release: make(chan struct{})} + healthy := &fakeSink{} + done := make(chan error, 1) + go func() { + done <- s.fanOut(&fakeProcessor{events: []*flowctlv1.Event{{Id: "1", Type: "test.event"}}}, "processor", []sinkStream{{id: "slow", stream: slow}, {id: "healthy", stream: healthy}}) + }() + <-slow.blocked + select { + case err := <-done: + t.Fatalf("fan-out completed while sink was blocked: %v", err) + default: + } + close(slow.release) + if err := <-done; err != nil { + t.Fatal(err) + } + if slow.sent != 1 || healthy.sent != 1 { + t.Fatalf("lost event: slow=%d healthy=%d", slow.sent, healthy.sent) + } +} + +func TestWaitCancelsOtherEdges(t *testing.T) { + s := testOrchestrator(t) + failure := errors.New("source failed") + exited := make(chan struct{}) + s.launch(func() error { <-s.ctx.Done(); close(exited); return s.ctx.Err() }) + s.launch(func() error { return s.edge("source", "processor", "source-processor", "recv", nil, failure) }) + if err := s.Wait(); !errors.Is(err, failure) { + t.Fatalf("wait = %v", err) + } + <-exited +} diff --git a/schemas/cue/schema.cue b/schemas/cue/schema.cue index 6129bb0..92313f4 100644 --- a/schemas/cue/schema.cue +++ b/schemas/cue/schema.cue @@ -39,6 +39,9 @@ package pipeline // Optional driver for deployment driver?: "docker" | "kubernetes" | "k8s" | "nomad" | "local" + // Explicit opt-out for components that only register with the control plane. + registrationOnly?: bool + // Sources - can be empty if pipelines are present sources: [...#SourceComponent] diff --git a/test/delivery_process_test.go b/test/delivery_process_test.go new file mode 100644 index 0000000..74d85cf --- /dev/null +++ b/test/delivery_process_test.go @@ -0,0 +1,146 @@ +package test + +import ( + "bytes" + "context" + "fmt" + "github.com/withobsrvr/flowctl/internal/storage" + flowctlpb "github.com/withobsrvr/flowctl/proto" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" +) + +// Exercises the CLI, process driver, control-plane registration, and gRPC +// forwarding with a source that returns a non-EOF error across the wire. +func TestProcessDeliveryTerminalResult(t *testing.T) { + if testing.Short() { + t.Skip("process-driver integration") + } + root := findRepoRoot(t) + dir := t.TempDir() + flowctl := filepath.Join(dir, "flowctl") + fixture := filepath.Join(dir, "delivery-component") + buildGoBinary(t, root, flowctl, ".") + buildGoBinary(t, root, fixture, filepath.Join(root, "test", "testdata", "delivery_component.go")) + for _, tc := range []struct { + mode, processorMode, sinkMode string + bounded, failed, direct bool + want, edge string + }{ + {"fail-before", "", "", false, true, false, "synthetic source failure", "source -> processor"}, + {"fail-after", "", "", false, true, false, "synthetic source failure", "source -> processor"}, + {"bounded", "", "", true, false, false, "RUN_STATUS_COMPLETED", ""}, + {"unbounded-eof", "", "", false, true, false, "unexpected EOF", "source -> processor"}, + {"bounded", "processor-fail", "", true, true, false, "synthetic processor failure", "processor -> sinks"}, + {"bounded", "", "sink-fail", true, true, false, "synthetic sink failure", "processor -> sink"}, + {"wait", "", "", false, false, false, "RUN_STATUS_STOPPED", ""}, + {"fail-before", "", "", false, true, true, "synthetic source failure", "source -> sinks"}, + {"fail-after", "", "", false, true, true, "synthetic source failure", "source -> sinks"}, + {"unbounded-eof", "", "", false, true, true, "unexpected EOF", "source -> sinks"}, + {"bounded", "", "", true, false, true, "RUN_STATUS_COMPLETED", ""}, + {"bounded", "", "sink-fail", true, true, true, "synthetic sink failure", "source -> sink"}, + } { + t.Run(fmt.Sprintf("%s-%s-%s-direct=%t", tc.mode, tc.processorMode, tc.sinkMode, tc.direct), func(t *testing.T) { + sourceEnv := fmt.Sprintf(" DELIVERY_MODE: %s\n", tc.mode) + if tc.bounded { + sourceEnv += " END_LEDGER: \"2\"\n" + } + processorSection := fmt.Sprintf(` processors: + - id: processor + command: [%q] + inputs: [source] + env: + DELIVERY_ROLE: processor + DELIVERY_MODE: %q +`, fixture, tc.processorMode) + sinkInput := "processor" + if tc.direct { + processorSection = "" + sinkInput = "source" + } + pipeline := fmt.Sprintf(`apiVersion: flowctl/v1 +kind: Pipeline +metadata: + name: delivery-test +spec: + driver: process + sources: + - id: source + command: [%q] + env: + DELIVERY_ROLE: source +%s%s sinks: + - id: sink + command: [%q] + inputs: [%s] + env: + DELIVERY_ROLE: sink + DELIVERY_MODE: %q +`, fixture, sourceEnv, processorSection, fixture, sinkInput, tc.sinkMode) + file := filepath.Join(dir, strings.ReplaceAll(t.Name(), "/", "-")+".yaml") + db := filepath.Join(dir, strings.ReplaceAll(t.Name(), "/", "-")+".db") + if err := os.WriteFile(file, []byte(pipeline), 0600); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, flowctl, "run", "--show-status=false", "--db-path", db, "--control-plane-port=0", "--log-dir", filepath.Join(dir, tc.mode+"-logs"), file) + cmd.Dir = root + var out []byte + var err error + if tc.mode == "wait" { + var buf bytes.Buffer + cmd.Stdout, cmd.Stderr = &buf, &buf + if err = cmd.Start(); err != nil { + t.Fatal(err) + } + time.Sleep(2 * time.Second) + if err = cmd.Process.Signal(os.Interrupt); err != nil { + t.Fatal(err) + } + err = cmd.Wait() + out = buf.Bytes() + } else { + out, err = cmd.CombinedOutput() + } + if ctx.Err() != nil { + t.Fatalf("run timed out: %s", out) + } + if tc.failed && err == nil { + t.Fatalf("expected nonzero CLI exit: %s", out) + } + if !tc.failed && err != nil { + t.Fatalf("expected successful completion: %v\n%s", err, out) + } + if !strings.Contains(string(out), tc.want) { + t.Fatalf("missing %q in output:\n%s", tc.want, out) + } + if tc.failed && !strings.Contains(string(out), tc.edge) { + t.Fatalf("missing failing edge:\n%s", out) + } + if tc.failed && !strings.Contains(string(out), "RUN_STATUS_FAILED") { + t.Fatalf("run was not marked failed:\n%s", out) + } + store := storage.NewBoltDBStorage(&storage.BoltOptions{Path: db}) + if err := store.Open(); err != nil { + t.Fatal(err) + } + runs, err := store.ListPipelineRuns(context.Background(), "delivery-test", flowctlpb.RunStatus_RUN_STATUS_UNKNOWN, 10) + _ = store.Close() + if err != nil || len(runs) != 1 { + t.Fatalf("stored runs=%d err=%v", len(runs), err) + } + if tc.failed { + if (!strings.Contains(runs[0].Run.Error, `"from":"source"`) && !strings.Contains(runs[0].Run.Error, `"from":"processor"`)) || !strings.Contains(runs[0].Run.Error, `"cause":`) || strings.Contains(runs[0].Run.Error, "synthetic source failure") { + t.Fatalf("missing or unsafe persisted structured failure: %s", runs[0].Run.Error) + } + } else if runs[0].Run.Error != "" { + t.Fatalf("unexpected persisted error: %s", runs[0].Run.Error) + } + }) + } +} diff --git a/test/direct_sink_routing_test.go b/test/direct_sink_routing_test.go new file mode 100644 index 0000000..7dbc775 --- /dev/null +++ b/test/direct_sink_routing_test.go @@ -0,0 +1,84 @@ +package test + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" +) + +// Each source must reach only sinks which declare it in inputs. Event-type +// filtering alone is insufficient when both sources emit the same type. +func TestProcessDirectSinkInputs(t *testing.T) { + if testing.Short() { + t.Skip("process-driver integration") + } + root := findRepoRoot(t) + dir := t.TempDir() + flowctl := filepath.Join(dir, "flowctl") + fixture := filepath.Join(dir, "delivery-component") + buildGoBinary(t, root, flowctl, ".") + buildGoBinary(t, root, fixture, filepath.Join(root, "test", "testdata", "delivery_component.go")) + pipeline := fmt.Sprintf(`apiVersion: flowctl/v1 +kind: Pipeline +metadata: + name: direct-routing +spec: + driver: process + sources: + - id: source-a + command: [%q] + env: + DELIVERY_ROLE: source + DELIVERY_MODE: bounded + END_LEDGER: "2" + - id: source-b + command: [%q] + env: + DELIVERY_ROLE: source + DELIVERY_MODE: bounded + END_LEDGER: "2" + sinks: + - id: sink-a + command: [%q] + inputs: [source-a] + env: + DELIVERY_ROLE: sink + - id: sink-b + command: [%q] + inputs: [source-b] + env: + DELIVERY_ROLE: sink +`, fixture, fixture, fixture, fixture) + file := filepath.Join(dir, "pipeline.yaml") + if err := os.WriteFile(file, []byte(pipeline), 0600); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, flowctl, "run", "--no-persistence", "--show-status=false", "--control-plane-port=0", file) + cmd.Dir = dir // process driver writes component logs to its working directory + out, err := cmd.CombinedOutput() + if err != nil || ctx.Err() != nil { + t.Fatalf("pipeline did not complete: %v\n%s", err, out) + } + if !strings.Contains(string(out), "RUN_STATUS_COMPLETED") { + t.Fatalf("run not completed:\n%s", out) + } + for _, tc := range []struct{ sink, want, reject string }{ + {"sink-a", "source-a", "source-b"}, + {"sink-b", "source-b", "source-a"}, + } { + contents, err := os.ReadFile(filepath.Join(dir, "logs", tc.sink+".log")) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(contents), "DELIVERY_RECEIVED id="+tc.want) || strings.Contains(string(contents), "DELIVERY_RECEIVED id="+tc.reject) { + t.Fatalf("%s received wrong source:\n%s", tc.sink, contents) + } + } +} diff --git a/test/operator_flow_test.go b/test/operator_flow_test.go index 62a1981..26086f6 100644 --- a/test/operator_flow_test.go +++ b/test/operator_flow_test.go @@ -40,6 +40,7 @@ metadata: name: operator-smoke spec: driver: process + registrationOnly: true sources: - id: test-source command: [%q] diff --git a/test/testdata/delivery_component.go b/test/testdata/delivery_component.go new file mode 100644 index 0000000..0b5c490 --- /dev/null +++ b/test/testdata/delivery_component.go @@ -0,0 +1,140 @@ +// A standalone process-driver fixture for delivery failure and completion tests. +package main + +import ( + "context" + "io" + "log" + "net" + "os" + "os/signal" + "syscall" + "time" + + flowctlv1 "github.com/withObsrvr/flow-proto/go/gen/flowctl/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" +) + +type sourceService struct { + flowctlv1.UnimplementedSourceServiceServer +} +type processorService struct { + flowctlv1.UnimplementedProcessorServiceServer +} +type sinkService struct { + flowctlv1.UnimplementedConsumerServiceServer +} + +func (sourceService) StreamEvents(_ *flowctlv1.StreamRequest, stream grpc.ServerStreamingServer[flowctlv1.Event]) error { + if os.Getenv("DELIVERY_MODE") != "fail-before" { + if err := stream.Send(&flowctlv1.Event{Id: os.Getenv("FLOWCTL_COMPONENT_ID"), Type: "test.event"}); err != nil { + return err + } + } + switch os.Getenv("DELIVERY_MODE") { + case "fail-before", "fail-after": + return status.Error(codes.Internal, "synthetic source failure") + case "unbounded-eof", "bounded": + return nil + default: + <-stream.Context().Done() + return stream.Context().Err() + } +} +func (processorService) Process(stream grpc.BidiStreamingServer[flowctlv1.Event, flowctlv1.Event]) error { + if os.Getenv("DELIVERY_MODE") == "processor-fail" { + return status.Error(codes.Internal, "synthetic processor failure") + } + for { + event, err := stream.Recv() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + if err := stream.Send(event); err != nil { + return err + } + } +} +func (sinkService) Consume(stream grpc.ClientStreamingServer[flowctlv1.Event, flowctlv1.ConsumeResponse]) error { + if os.Getenv("DELIVERY_MODE") == "sink-fail" { + return status.Error(codes.Internal, "synthetic sink failure") + } + for { + event, err := stream.Recv() + if err == nil { + log.Printf("DELIVERY_RECEIVED id=%s", event.Id) + } + if err == io.EOF { + return stream.SendAndClose(&flowctlv1.ConsumeResponse{}) + } + if err != nil { + return err + } + } +} +func main() { + id, role, endpoint := os.Getenv("FLOWCTL_COMPONENT_ID"), os.Getenv("DELIVERY_ROLE"), os.Getenv("FLOWCTL_ENDPOINT") + if id == "" || role == "" || endpoint == "" { + log.Fatal("missing fixture environment") + } + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + log.Fatal(err) + } + server := grpc.NewServer() + kind := flowctlv1.ComponentType_COMPONENT_TYPE_SOURCE + switch role { + case "source": + flowctlv1.RegisterSourceServiceServer(server, sourceService{}) + case "processor": + kind = flowctlv1.ComponentType_COMPONENT_TYPE_PROCESSOR + flowctlv1.RegisterProcessorServiceServer(server, processorService{}) + case "sink": + kind = flowctlv1.ComponentType_COMPONENT_TYPE_CONSUMER + flowctlv1.RegisterConsumerServiceServer(server, sinkService{}) + default: + log.Fatal("unknown fixture role") + } + go func() { + if err := server.Serve(listener); err != nil { + log.Print(err) + } + }() + conn, err := grpc.NewClient(endpoint, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + log.Fatal(err) + } + defer conn.Close() + client := flowctlv1.NewControlPlaneServiceClient(conn) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + resp, err := client.RegisterComponent(ctx, &flowctlv1.RegisterRequest{ComponentId: id, Component: &flowctlv1.ComponentInfo{Id: id, Name: id, Type: kind, Endpoint: listener.Addr().String(), InputEventTypes: []string{"test.event"}, OutputEventTypes: []string{"test.event"}}}) + if err != nil { + log.Fatal(err) + } + heartbeat := func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, _ = client.Heartbeat(ctx, &flowctlv1.HeartbeatRequest{ServiceId: resp.ServiceId, Status: flowctlv1.HealthStatus_HEALTH_STATUS_HEALTHY}) + } + heartbeat() + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGTERM, os.Interrupt) + defer signal.Stop(sig) + for { + select { + case <-ticker.C: + heartbeat() + case <-sig: + return + } + } +}