[cmd/opampsupervisor] Control Collectors with only the OpAMP extension

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: []
+
+# (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/supervisor/commander/commander.go

@@ -201,6 +201,102 @@ func (c *Commander) watch() {
 	c.exitCh <- struct{}{}
 }
 
+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(stdout, 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() {
+		cmd.Wait()
+		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

cmd/opampsupervisor/supervisor/supervisor.go

@@ -4,6 +4,7 @@
 package supervisor
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"crypto/tls"
@@ -16,6 +17,7 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"text/template"
@@ -73,8 +75,9 @@ var (
 )
 
 const (
-	persistentStateFileName = "persistent_state.yaml"
-	agentConfigFileName     = "effective.yaml"
+	persistentStateFileName     = "persistent_state.yaml"
+	agentConfigFileName         = "effective.yaml"
+	AllowNoPipelinesFeatureGate = "service.AllowNoPipelines"
 )
 
 const maxBufferedCustomMessages = 10
@@ -182,6 +185,8 @@ type Supervisor struct {
 	opampServerPort int
 
 	telemetrySettings telemetrySettings
+
+	featureGates map[string]struct{}
 }
 
 func NewSupervisor(logger *zap.Logger, cfg config.Supervisor) (*Supervisor, error) {
@@ -196,6 +201,7 @@ func NewSupervisor(logger *zap.Logger, cfg config.Supervisor) (*Supervisor, erro
 		doneChan:                     make(chan struct{}),
 		customMessageToServer:        make(chan *protobufs.CustomMessage, maxBufferedCustomMessages),
 		agentConn:                    &atomic.Value{},
+		featureGates:                 map[string]struct{}{},
 	}
 	if err := s.createTemplates(); err != nil {
 		return nil, err
@@ -312,6 +318,10 @@ func (s *Supervisor) Start() error {
 		return err
 	}
 
+	if err = s.getFeatureGates(); err != nil {
+		return fmt.Errorf("could not get feature gates from the Collector: %w", err)
+	}
+
 	if err = s.getBootstrapInfo(); err != nil {
 		return fmt.Errorf("could not get bootstrap info from the Collector: %w", err)
 	}
@@ -365,6 +375,41 @@ func (s *Supervisor) Start() error {
 	return nil
 }
 
+func (s *Supervisor) getFeatureGates() error {
+	cmd, err := commander.NewCommander(
+		s.telemetrySettings.Logger,
+		s.config.Storage.Directory,
+		s.config.Agent,
+		"featuregate",
+	)
+	if err != nil {
+		return err
+	}
+
+	stdout, _, err := cmd.StartOneShot()
+	if err != nil {
+		return err
+	}
+	scanner := bufio.NewScanner(bytes.NewBuffer(stdout))
+
+	// First line only contains headers, discard it.
+	_ = scanner.Scan()
+	for scanner.Scan() {
+		line := scanner.Text()
+		i := strings.Index(line, " ")
+		flag := line[0:i]
+
+		if flag == AllowNoPipelinesFeatureGate {
+			s.featureGates[AllowNoPipelinesFeatureGate] = struct{}{}
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		fmt.Fprintln(os.Stderr, "reading standard input:", err)
+	}
+
+	return nil
+}
+
 func (s *Supervisor) createTemplates() error {
 	var err error
 
@@ -498,11 +543,18 @@ func (s *Supervisor) getBootstrapInfo() (err error) {
 		}
 	}()
 
+	flags := []string{
+		"--config", s.agentConfigFilePath(),
+	}
+	featuregateFlag := s.getFeatureGateFlag()
+	if len(featuregateFlag) > 0 {
+		flags = append(flags, featuregateFlag...)
+	}
 	cmd, err := commander.NewCommander(
 		s.telemetrySettings.Logger,
 		s.config.Storage.Directory,
 		s.config.Agent,
-		"--config", s.agentConfigFilePath(),
+		flags...,
 	)
 	if err != nil {
 		span.SetStatus(codes.Error, fmt.Sprintf("Could not start Agent: %v", err))
@@ -902,12 +954,12 @@ func (s *Supervisor) onOpampConnectionSettings(_ context.Context, settings *prot
 
 func (s *Supervisor) composeNoopPipeline() ([]byte, error) {
 	var cfg bytes.Buffer
-	err := s.noopPipelineTemplate.Execute(&cfg, map[string]any{
-		"InstanceUid":    s.persistentState.InstanceID.String(),
-		"SupervisorPort": s.opampServerPort,
-	})
-	if err != nil {
-		return nil, err
+
+	if !s.isFeatureGateSupported(AllowNoPipelinesFeatureGate) {
+		err := s.noopPipelineTemplate.Execute(&cfg, map[string]any{})
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	return cfg.Bytes(), nil
@@ -1664,6 +1716,24 @@ func (s *Supervisor) getSupervisorOpAMPServerPort() (int, error) {
 	return s.findRandomPort()
 }
 
+func (s *Supervisor) getFeatureGateFlag() []string {
+	flags := []string{}
+	for k, _ := range s.featureGates {
+		flags = append(flags, k)
+	}
+
+	if len(flags) == 0 {
+		return []string{}
+	}
+
+	return []string{"--feature-gates", strings.Join(flags, ",")}
+}
+
+func (s *Supervisor) isFeatureGateSupported(gate string) bool {
+	_, ok := s.featureGates[gate]
+	return ok
+}
+
 func (s *Supervisor) findRandomPort() (int, error) {
 	l, err := net.Listen("tcp", "localhost:0")
 	if err != nil {