flow: always allow running with a partially evaluated graph

cmd/internal/flowmode/cmd_run.go

@@ -258,6 +258,9 @@ func (fr *flowRun) Run(configFile string) error {
 	// Perform the initial reload. This is done after starting the HTTP server so
 	// that /metric and pprof endpoints are available while the Flow controller
 	// is loading.
+	//
+	// TODO(rfratto): add a flag to permit the initial load failing, allowing
+	// Flow to run in a partially-healthy state.
 	if err := reload(); err != nil {
 		var diags diag.Diagnostics
 		if errors.As(err, &diags) {

component/registry.go

@@ -167,6 +167,12 @@ type Registration struct {
 
 	// Build should construct a new component from an initial Arguments and set
 	// of options.
+	//
+	// If Build returns an error, downstream components which depend on this
+	// component will not be evaluated.
+	//
+	// If Build returns a non-nil Component instance, that Component instance
+	// will run regardless of whether Build also returns a non-nil error.
 	Build func(opts Options, args Arguments) (Component, error)
 }
 

pkg/flow/flow.go

@@ -247,6 +247,28 @@ func (f *Flow) Run(ctx context.Context) {
 				f.loader.EvaluateDependencies(updated)
 			}
 
+			// If the graph was partially evaluated (i.e., one failed component
+			// prevented downstream components from initially evaluating), it's
+			// possible for a subset of components to not be running.
+			//
+			// We'll re-synchronize the list of running components to ensure that
+			// newly evaluated components are started.
+			//
+			// TODO(rfratto): this may be expensive with busy graphs; one solution
+			// could be to check for non-running components before calling
+			// Synchronize, or to check for a difference in the length of
+			// current synchronized runnables to the current list of
+			// f.loader.Components().
+			components := f.loader.Components()
+			runnables := make([]controller.RunnableNode, 0, len(components))
+			for _, uc := range components {
+				runnables = append(runnables, uc)
+			}
+			err := f.sched.Synchronize(runnables)
+			if err != nil {
+				level.Error(f.log).Log("msg", "failed to load components", "err", err)
+			}
+
 		case <-f.loadFinished:
 			level.Info(f.log).Log("msg", "scheduling loaded components and services")
 
@@ -280,19 +302,15 @@ func (f *Flow) Run(ctx context.Context) {
 // file. Components in the graph will be marked as unhealthy if there was an
 // error encountered during Load.
 //
-// The controller will only start running components after Load is called once
-// without any configuration errors.
+// If the Flow controller is running, loaded components will be scheduled for
+// running.
 func (f *Flow) LoadFile(file *File, args map[string]any) error {
+	defer f.loadedOnce.Store(true)
+
 	f.loadMut.Lock()
 	defer f.loadMut.Unlock()
 
 	diags := f.loader.Apply(args, file.Components, file.ConfigBlocks)
-	if !f.loadedOnce.Load() && diags.HasErrors() {
-		// The first call to Load should not run any components if there were
-		// errors in the configuration file.
-		return diags
-	}
-	f.loadedOnce.Store(true)
 
 	select {
 	case f.loadFinished <- struct{}{}:
@@ -302,7 +320,7 @@ func (f *Flow) LoadFile(file *File, args map[string]any) error {
 	return diags.ErrorOrNil()
 }
 
-// Ready returns whether the Flow controller has finished its initial load.
+// Ready returns whether LoadFile has been invoked at least once.
 func (f *Flow) Ready() bool {
 	return f.loadedOnce.Load()
 }

pkg/flow/internal/controller/node_component.go

@@ -284,12 +284,14 @@ func (cn *ComponentNode) evaluate(scope *vm.Scope) error {
 	if cn.managed == nil {
 		// We haven't built the managed component successfully yet.
 		managed, err := cn.reg.Build(cn.managedOpts, argsCopyValue)
-		if err != nil {
-			return fmt.Errorf("building component: %w", err)
-		}
 		cn.managed = managed
 		cn.args = argsCopyValue
 
+		// We do the error check at the end to allow the component to still return
+		// a component instance with an error to signal that it should still run.
+		if err != nil {
+			return fmt.Errorf("building component: %w", err)
+		}
 		return nil
 	}