[cmd/opampsupervisor] Control Collectors with only the OpAMP extension (#38809)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue. Ex. Adding a feature - Explain what this achieves.--> #### Description Right now three components are required for the Supervisor to control a Collector: * nopreceiver * nopexporter * opampextension The `service.AllowNoPipelines` feature gate allows the Collector to be started without pipelines, which was released in v0.122.0. This can be used to start the Collector with only the OpAMP extension during the bootstrapping process, obviating the need for the nopreceiver and nopexporter. --------- Co-authored-by: Evan Bradley <evan-bradley@users.noreply.github.com>

Author: evan-bradley

.chloggen/supervisor-no-pipelines.yaml

@@ -0,0 +1,30 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: opampsupervisor
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Allow controlling Collectors that don't include the nopreceiver and nopexporer
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38809]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+ This requires Collectors built with Collector API v0.122.0+. The nopreceiver
+ and nopexporter will continue to be supported for a few releases, after which
+ only v0.122.0+ will be supported.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

cmd/opampsupervisor/e2e_test.go

@@ -617,77 +617,109 @@ func TestSupervisorConfiguresCapabilities(t *testing.T) {
 }
 
 func TestSupervisorBootstrapsCollector(t *testing.T) {
-agentDescription := atomic.Value{}
+tests := []struct {
+name string
+cfg string
+env []string
+precheck func(t *testing.T)
+}{
+{
+name: "With service.AllowNoPipelines",
+cfg: "nocap",
+},
+{
+name: "Without service.AllowNoPipelines",
+cfg: "no_fg",
+env: []string{
+"COLLECTOR_BIN=../../bin/otelcontribcol_" + runtime.GOOS + "_" + runtime.GOARCH,
+},
+precheck: func(t *testing.T) {
+if runtime.GOOS == "windows" {
+t.Skip("This test requires a shell script, which may not be supported by Windows")
+}
+},
+},
+}
 
-// Load the Supervisor config so we can get the location of
-// the Collector that will be run.
-var cfg config.Supervisor
-cfgFile := getSupervisorConfig(t, "nocap", map[string]string{})
-k := koanf.New("::")
-err := k.Load(file.Provider(cfgFile.Name()), yaml.Parser())
-require.NoError(t, err)
-err = k.UnmarshalWithConf("", &cfg, koanf.UnmarshalConf{
-Tag: "mapstructure",
-})
-require.NoError(t, err)
+for _, tt := range tests {
+t.Run(tt.name, func(t *testing.T) {
+agentDescription := atomic.Value{}
 
-// Get the binary name and version from the Collector binary
-// using the `components` command that prints a YAML-encoded
-// map of information about the Collector build. Some of this
-// information will be used as defaults for the telemetry
-// attributes.
-agentPath := cfg.Agent.Executable
-componentsInfo, err := exec.Command(agentPath, "components").Output()
-require.NoError(t, err)
-k = koanf.New("::")
-err = k.Load(rawbytes.Provider(componentsInfo), yaml.Parser())
-require.NoError(t, err)
-buildinfo := k.StringMap("buildinfo")
-command := buildinfo["command"]
-version := buildinfo["version"]
+// Load the Supervisor config so we can get the location of
+// the Collector that will be run.
+var cfg config.Supervisor
+cfgFile := getSupervisorConfig(t, tt.cfg, map[string]string{})
+k := koanf.New("::")
+err := k.Load(file.Provider(cfgFile.Name()), yaml.Parser())
+require.NoError(t, err)
+err = k.UnmarshalWithConf("", &cfg, koanf.UnmarshalConf{
+Tag: "mapstructure",
+})
+require.NoError(t, err)
 
-server := newOpAMPServer(
-t,
-defaultConnectingHandler,
-types.ConnectionCallbacks{
-OnMessage: func(_ context.Context, _ types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
-if message.AgentDescription != nil {
-agentDescription.Store(message.AgentDescription)
-}
+// Get the binary name and version from the Collector binary
+// using the `components` command that prints a YAML-encoded
+// map of information about the Collector build. Some of this
+// information will be used as defaults for the telemetry
+// attributes.
+agentPath := cfg.Agent.Executable
+cmd := exec.Command(agentPath, "components")
+for _, env := range tt.env {
+cmd.Env = append(cmd.Env, env)
+}
+componentsInfo, err := cmd.Output()
+require.NoError(t, err)
+k = koanf.New("::")
+err = k.Load(rawbytes.Provider(componentsInfo), yaml.Parser())
+require.NoError(t, err)
+buildinfo := k.StringMap("buildinfo")
+command := buildinfo["command"]
+version := buildinfo["version"]
+
+server := newOpAMPServer(
+t,
+defaultConnectingHandler,
+types.ConnectionCallbacks{
+OnMessage: func(_ context.Context, _ types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
+if message.AgentDescription != nil {
+agentDescription.Store(message.AgentDescription)
+}
 
-return &protobufs.ServerToAgent{}
-},
-})
+return &protobufs.ServerToAgent{}
+},
+})
 
-s := newSupervisor(t, "nocap", map[string]string{"url": server.addr})
+s := newSupervisor(t, "nocap", map[string]string{"url": server.addr})
 
-require.Nil(t, s.Start())
-defer s.Shutdown()
+require.Nil(t, s.Start())
+defer s.Shutdown()
 
-waitForSupervisorConnection(server.supervisorConnected, true)
+waitForSupervisorConnection(server.supervisorConnected, true)
 
-require.Eventually(t, func() bool {
-ad, ok := agentDescription.Load().(*protobufs.AgentDescription)
-if !ok {
-return false
-}
+require.Eventually(t, func() bool {
+ad, ok := agentDescription.Load().(*protobufs.AgentDescription)
+if !ok {
+return false
+}
 
-var agentName, agentVersion string
-identAttr := ad.IdentifyingAttributes
-for _, attr := range identAttr {
-switch attr.Key {
-case semconv.AttributeServiceName:
-agentName = attr.Value.GetStringValue()
-case semconv.AttributeServiceVersion:
-agentVersion = attr.Value.GetStringValue()
-}
-}
+var agentName, agentVersion string
+identAttr := ad.IdentifyingAttributes
+for _, attr := range identAttr {
+switch attr.Key {
+case semconv.AttributeServiceName:
+agentName = attr.Value.GetStringValue()
+case semconv.AttributeServiceVersion:
+agentVersion = attr.Value.GetStringValue()
+}
+}
 
-// By default the Collector should report its name and version
-// from the component.BuildInfo struct built into the Collector
-// binary.
-return agentName == command && agentVersion == version
-}, 5*time.Second, 250*time.Millisecond)
+// By default the Collector should report its name and version
+// from the component.BuildInfo struct built into the Collector
+// binary.
+return agentName == command && agentVersion == version
+}, 5*time.Second, 250*time.Millisecond)
+})
+}
 }
 
 func TestSupervisorBootstrapsCollectorAvailableComponents(t *testing.T) {

cmd/opampsupervisor/supervisor/commander/commander.go

@@ -201,6 +201,108 @@ func (c *Commander) watch() {
 c.exitCh <- struct{}{}
 }
 
+// StartOneShot starts the Collector with the expectation that it will immediately
+// exit after it finishes a quick operation. This is useful for situations like reading stdout/sterr
+// to e.g. check the feature gate the Collector supports.
+func (c *Commander) StartOneShot() ([]byte, []byte, error) {
+stdout := []byte{}
+stderr := []byte{}
+ctx := context.Background()
+
+cmd := exec.CommandContext(ctx, c.cfg.Executable, c.args...) // #nosec G204
+cmd.Env = common.EnvVarMapToEnvMapSlice(c.cfg.Env)
+cmd.SysProcAttr = sysProcAttrs()
+// grab cmd pipes
+stdoutPipe, err := cmd.StdoutPipe()
+if err != nil {
+return nil, nil, fmt.Errorf("stdoutPipe: %w", err)
+}
+stderrPipe, err := cmd.StderrPipe()
+if err != nil {
+return nil, nil, fmt.Errorf("stderrPipe: %w", err)
+}
+
+// start agent
+if err := cmd.Start(); err != nil {
+return nil, nil, fmt.Errorf("start: %w", err)
+}
+// capture agent output
+go func() {
+scanner := bufio.NewScanner(stdoutPipe)
+for scanner.Scan() {
+stdout = append(stdout, scanner.Bytes()...)
+stdout = append(stdout, byte('\n'))
+}
+if err := scanner.Err(); err != nil {
+c.logger.Error("Error reading agent stdout: %w", zap.Error(err))
+}
+}()
+go func() {
+scanner := bufio.NewScanner(stderrPipe)
+for scanner.Scan() {
+stderr = append(stderr, scanner.Bytes()...)
+stderr = append(stderr, byte('\n'))
+}
+if err := scanner.Err(); err != nil {
+c.logger.Error("Error reading agent stderr: %w", zap.Error(err))
+}
+}()
+
+c.logger.Debug("Agent process started", zap.Int("pid", cmd.Process.Pid))
+
+doneCh := make(chan struct{}, 1)
+
+go func() {
+err := cmd.Wait()
+if err != nil {
+c.logger.Error("One-shot Collector encountered an error during execution", zap.Error(err))
+}
+doneCh <- struct{}{}
+}()
+
+waitCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
+
+defer cancel()
+
+select {
+case <-doneCh:
+case <-waitCtx.Done():
+pid := cmd.Process.Pid
+c.logger.Debug("Stopping agent process", zap.Int("pid", pid))
+
+// Gracefully signal process to stop.
+if err := sendShutdownSignal(cmd.Process); err != nil {
+return nil, nil, err
+}
+
+innerWaitCtx, innerCancel := context.WithTimeout(ctx, 10*time.Second)
+
+// Setup a goroutine to wait a while for process to finish and send kill signal
+// to the process if it doesn't finish.
+var innerErr error
+go func() {
+<-innerWaitCtx.Done()
+
+if !errors.Is(innerWaitCtx.Err(), context.DeadlineExceeded) {
+c.logger.Debug("Agent process successfully stopped.", zap.Int("pid", pid))
+return
+}
+
+// Time is out. Kill the process.
+c.logger.Debug(
+"Agent process is not responding to SIGTERM. Sending SIGKILL to kill forcibly.",
+zap.Int("pid", pid))
+if innerErr = cmd.Process.Signal(os.Kill); innerErr != nil {
+return
+}
+}()
+
+innerCancel()
+}
+
+return stdout, stderr, nil
+}
+
 // Exited returns a channel that will send a signal when the Agent process exits.
 func (c *Commander) Exited() <-chan struct{} {
 return c.exitCh