Introduce isolated units for input spec (#4476)

* Introduce non grouped input components

* Create fragment

* Edit fragment

* Additional comments

* Fake test grouped

* Additional tests

* Addition of non grouped fake type

* Removing unused

* Fix case

* Rename flag

* Rename label

* State change

* Attempt without shipper

* Trying output without shipper

* Revert "Trying output without shipper"

This reverts commit 62d3e93.

* Revert "Attempt without shipper"

This reverts commit fddaad6.

* Trying single unit

* Testing windows multi shipper inputs on same npipe

* Rename and disable isolated on windows

* Reduce duplication of code

* Enable on windows and refactor grpc servers

* Additional comments

* Remove windows comments

Author: jeniawhite

changelog/fragments/1711355973-Introduce-non-grouping-for-input-spec.yaml

@@ -0,0 +1,34 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Introduce isolate units for input spec.
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+description: |
+  Introduce a new flag in the spec that will allow to disable grouping of same input type inputs into a single component.
+  If that flag is being activated the inputs won't be grouped and we will create a separate component for each input that will run units for that particular input.
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/elastic-agent/pull/4476
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/security-team/issues/8669

internal/pkg/agent/application/coordinator/coordinator_test.go

@@ -61,6 +61,19 @@ var (
 			},
 		},
 	}
+	fakeIsolatedUnitsInputSpec = component.InputSpec{
+		Name:      "fake-isolated-units",
+		Platforms: []string{fmt.Sprintf("%s/%s", goruntime.GOOS, goruntime.GOARCH)},
+		Shippers:  []string{"fake-shipper"},
+		Command: &component.CommandSpec{
+			Timeouts: component.CommandTimeoutSpec{
+				Checkin: 30 * time.Second,
+				Restart: 10 * time.Millisecond, // quick restart during tests
+				Stop:    30 * time.Second,
+			},
+		},
+		IsolateUnits: true,
+	}
 	fakeShipperSpec = component.ShipperSpec{
 		Name:      "fake-shipper",
 		Platforms: []string{fmt.Sprintf("%s/%s", goruntime.GOOS, goruntime.GOARCH)},
@@ -547,6 +560,94 @@ func TestCoordinator_StateSubscribe(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestCoordinator_StateSubscribeIsolatedUnits(t *testing.T) {
+	coordCh := make(chan error)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	coord, cfgMgr, varsMgr := createCoordinator(t, ctx, WithComponentInputSpec(fakeIsolatedUnitsInputSpec))
+	go func() {
+		err := coord.Run(ctx)
+		if errors.Is(err, context.Canceled) {
+			// allowed error
+			err = nil
+		}
+		coordCh <- err
+	}()
+
+	resultChan := make(chan error)
+	go func() {
+		ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+		defer cancel()
+
+		subChan := coord.StateSubscribe(ctx, 32)
+		for {
+			select {
+			case <-ctx.Done():
+				resultChan <- ctx.Err()
+				return
+			case state := <-subChan:
+				t.Logf("%+v", state)
+				if len(state.Components) == 3 {
+					compState0 := getComponentState(state.Components, "fake-isolated-units-default-fake-isolated-units-0")
+					compState1 := getComponentState(state.Components, "fake-isolated-units-default-fake-isolated-units-1")
+					if compState0 != nil && compState1 != nil {
+						unit0, ok0 := compState0.State.Units[runtime.ComponentUnitKey{UnitType: client.UnitTypeInput, UnitID: "fake-isolated-units-default-fake-isolated-units-0-unit"}]
+						unit1, ok1 := compState1.State.Units[runtime.ComponentUnitKey{UnitType: client.UnitTypeInput, UnitID: "fake-isolated-units-default-fake-isolated-units-1-unit"}]
+						if ok0 && ok1 {
+							if (unit0.State == client.UnitStateHealthy && unit0.Message == "Healthy From Fake Isolated Units 0 Config") &&
+								(unit1.State == client.UnitStateHealthy && unit1.Message == "Healthy From Fake Isolated Units 1 Config") {
+								resultChan <- nil
+								return
+							}
+						}
+					}
+				}
+			}
+		}
+	}()
+
+	// no vars used by the config
+	varsMgr.Vars(ctx, []*transpiler.Vars{{}})
+
+	// set the configuration to run a fake input
+	cfg, err := config.NewConfigFrom(map[string]interface{}{
+		"outputs": map[string]interface{}{
+			"default": map[string]interface{}{
+				"type": "fake-action-output",
+				"shipper": map[string]interface{}{
+					"enabled": true,
+				},
+			},
+		},
+		"inputs": []interface{}{
+			map[string]interface{}{
+				"id":         "fake-isolated-units-0",
+				"type":       "fake-isolated-units",
+				"use_output": "default",
+				"state":      client.UnitStateHealthy,
+				"message":    "Healthy From Fake Isolated Units 0 Config",
+			},
+			map[string]interface{}{
+				"id":         "fake-isolated-units-1",
+				"type":       "fake-isolated-units",
+				"use_output": "default",
+				"state":      client.UnitStateHealthy,
+				"message":    "Healthy From Fake Isolated Units 1 Config",
+			},
+		},
+	})
+	require.NoError(t, err)
+	cfgMgr.Config(ctx, cfg)
+
+	err = <-resultChan
+	require.NoError(t, err)
+	cancel()
+
+	err = <-coordCh
+	require.NoError(t, err)
+}
+
 func TestCollectManagerErrorsTimeout(t *testing.T) {
 	handlerChan, _, _, _, _ := setupManagerShutdownChannels(time.Millisecond)
 	// Don't send anything to the shutdown channels, causing a timeout
@@ -757,6 +858,7 @@ func TestCoordinator_UpgradeDetails(t *testing.T) {
 type createCoordinatorOpts struct {
 	managed        bool
 	upgradeManager UpgradeManager
+	compInputSpec  component.InputSpec
 }
 
 type CoordinatorOpt func(o *createCoordinatorOpts)
@@ -773,13 +875,21 @@ func WithUpgradeManager(upgradeManager UpgradeManager) CoordinatorOpt {
 	}
 }
 
+func WithComponentInputSpec(spec component.InputSpec) CoordinatorOpt {
+	return func(o *createCoordinatorOpts) {
+		o.compInputSpec = spec
+	}
+}
+
 // createCoordinator creates a coordinator that using a fake config manager and a fake vars manager.
 //
 // The runtime specifications is set up to use both the fake component and fake shipper.
 func createCoordinator(t *testing.T, ctx context.Context, opts ...CoordinatorOpt) (*Coordinator, *fakeConfigManager, *fakeVarsManager) {
 	t.Helper()
 
-	o := &createCoordinatorOpts{}
+	o := &createCoordinatorOpts{
+		compInputSpec: fakeInputSpec,
+	}
 	for _, opt := range opts {
 		opt(o)
 	}
@@ -793,7 +903,7 @@ func createCoordinator(t *testing.T, ctx context.Context, opts ...CoordinatorOpt
 		InputType:  "fake",
 		BinaryName: "",
 		BinaryPath: testBinary(t, "component"),
-		Spec:       fakeInputSpec,
+		Spec:       o.compInputSpec,
 	}
 	shipperSpec := component.ShipperRuntimeSpec{
 		ShipperType: "fake-shipper",

pkg/component/component.go

@@ -351,18 +351,13 @@ func unitForShipperOutput(output outputI, id string, shipperType string) Unit {
 	}
 }
 
-// Collect all inputs of the given type going to the given output and return
-// the resulting Component. The returned Component may have no units if no
-// active inputs were found.
-func (r *RuntimeSpecs) componentForInputType(
-	inputType string,
+// createShipperReference creates a ShipperReference for the given output and input spec.
+func (r *RuntimeSpecs) createShipperReference(
 	output outputI,
-	featureFlags *features.Flags,
-	componentConfig *ComponentConfig,
-) Component {
-	componentID := fmt.Sprintf("%s-%s", inputType, output.name)
-
-	inputSpec, componentErr := r.GetInput(inputType)
+	inputSpec InputRuntimeSpec,
+	componentID string,
+	componentErr error,
+) (*ShipperReference, error) {
 	var shipperRef *ShipperReference
 	if componentErr == nil {
 		if output.shipperEnabled {
@@ -393,58 +388,133 @@ func (r *RuntimeSpecs) componentForInputType(
 			}
 		}
 	}
+
 	// If there's an error at this point we still proceed with assembling the
 	// policy into a component, we just attach the error to its Err field to
 	// indicate that it can't be run.
+	return shipperRef, componentErr
+}
 
-	var units []Unit
-	for _, input := range output.inputs[inputType] {
-		if input.enabled {
-			unitID := fmt.Sprintf("%s-%s", componentID, input.id)
-			units = append(units, unitForInput(input, unitID))
+// populateOutputUnitsForInput adds the output units to the given slice.
+func populateOutputUnitsForInput(
+	units *[]Unit,
+	output outputI,
+	componentID string,
+	componentErr error,
+	shipperRef *ShipperReference,
+) {
+	if shipperRef != nil {
+		// Shipper units are skipped if componentErr isn't nil, because in that
+		// case we generally don't have a valid shipper type to base it on.
+		if componentErr == nil {
+			*units = append(*units, unitForShipperOutput(output, componentID, shipperRef.ShipperType))
 		}
+	} else {
+		*units = append(*units, unitForOutput(output, componentID))
 	}
-	if len(units) > 0 {
-		if shipperRef != nil {
-			// Shipper units are skipped if componentErr isn't nil, because in that
-			// case we generally don't have a valid shipper type to base it on.
-			if componentErr == nil {
-				units = append(units,
-					unitForShipperOutput(output, componentID, shipperRef.ShipperType))
+}
+
+// Collect all inputs of the given type going to the given output and return
+// the resulting Components. The returned Components may have no units if no
+// active inputs were found.
+func (r *RuntimeSpecs) componentsForInputType(
+	inputType string,
+	output outputI,
+	featureFlags *features.Flags,
+	componentConfig *ComponentConfig,
+) []Component {
+	var components []Component
+	inputSpec, componentErr := r.GetInput(inputType)
+
+	// Treat as non isolated units component on error of reading the input spec
+	if componentErr != nil || !inputSpec.Spec.IsolateUnits {
+		componentID := fmt.Sprintf("%s-%s", inputType, output.name)
+		shipperRef, componentErr := r.createShipperReference(output, inputSpec, componentID, componentErr)
+
+		var units []Unit
+		for _, input := range output.inputs[inputType] {
+			if input.enabled {
+				unitID := fmt.Sprintf("%s-%s", componentID, input.id)
+				units = append(units, unitForInput(input, unitID))
 			}
-		} else {
-			units = append(units, unitForOutput(output, componentID))
+		}
+
+		if len(units) > 0 {
+			// Populate the output units for this component
+			populateOutputUnitsForInput(
+				&units,
+				output,
+				componentID,
+				componentErr,
+				shipperRef,
+			)
+		}
+
+		components = append(components, Component{
+			ID:         componentID,
+			Err:        componentErr,
+			InputSpec:  &inputSpec,
+			InputType:  inputType,
+			OutputType: output.outputType,
+			Units:      units,
+			Features:   featureFlags.AsProto(),
+			Component:  componentConfig.AsProto(),
+			ShipperRef: shipperRef,
+		})
+	} else {
+		for _, input := range output.inputs[inputType] {
+			// Units are being mapped to components, so we need a unique ID for each.
+			componentID := fmt.Sprintf("%s-%s-%s", inputType, output.name, input.id)
+			shipperRef, componentErr := r.createShipperReference(output, inputSpec, componentID, componentErr)
+
+			var units []Unit
+			if input.enabled {
+				unitID := fmt.Sprintf("%s-unit", componentID)
+				units = append(units, unitForInput(input, unitID))
+				// Populate the output units for this component
+				populateOutputUnitsForInput(
+					&units,
+					output,
+					componentID,
+					componentErr,
+					shipperRef,
+				)
+			}
+
+			components = append(components, Component{
+				ID:         componentID,
+				Err:        componentErr,
+				InputSpec:  &inputSpec,
+				InputType:  inputType,
+				OutputType: output.outputType,
+				Units:      units,
+				Features:   featureFlags.AsProto(),
+				Component:  componentConfig.AsProto(),
+				ShipperRef: shipperRef,
+			})
 		}
 	}
-	return Component{
-		ID:         componentID,
-		Err:        componentErr,
-		InputSpec:  &inputSpec,
-		InputType:  inputType,
-		OutputType: output.outputType,
-		Units:      units,
-		Features:   featureFlags.AsProto(),
-		Component:  componentConfig.AsProto(),
-		ShipperRef: shipperRef,
-	}
+	return components
 }
 
 func (r *RuntimeSpecs) componentsForOutput(output outputI, featureFlags *features.Flags, componentConfig *ComponentConfig) []Component {
 	var components []Component
 	shipperTypes := make(map[string]bool)
 	for inputType := range output.inputs {
 		// No need for error checking at this stage -- we are guaranteed
-		// to get a Component back. If there is an error that prevents it
+		// to get a Component/s back. If there is an error that prevents it/them
 		// from running then it will be in the Component's Err field and
-		// we will report it later. The only thing we skip is a component
+		// we will report it later. The only thing we skip is a component/s
 		// with no units.
-		component := r.componentForInputType(inputType, output, featureFlags, componentConfig)
-		if len(component.Units) > 0 {
-			if component.ShipperRef != nil {
-				// If this component uses a shipper, mark that shipper type as active
-				shipperTypes[component.ShipperRef.ShipperType] = true
+		typeComponents := r.componentsForInputType(inputType, output, featureFlags, componentConfig)
+		for _, component := range typeComponents {
+			if len(component.Units) > 0 {
+				if component.ShipperRef != nil {
+					// If this component uses a shipper, mark that shipper type as active
+					shipperTypes[component.ShipperRef.ShipperType] = true
+				}
+				components = append(components, component)
 			}
-			components = append(components, component)
 		}
 	}