cleanup

Author: VihasMakwana

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

@@ -173,9 +173,6 @@ type OTelManager interface {
 	// PerformComponentDiagnostics executes the diagnostic action for the provided components. If no components are provided,
 	// then it performs the diagnostics for all current units.
 	PerformComponentDiagnostics(ctx context.Context, additionalMetrics []cproto.AdditionalDiagnosticRequest, req ...component.Component) ([]runtime.ComponentDiagnostic, error)
-
-	// DiagnosticHooks returns the list of "global hooks" for the EDOT process.
-	DiagnosticHooks() diagnostics.Hooks
 }
 
 // ConfigChange provides an interface for receiving a new configuration.
@@ -1286,27 +1283,8 @@ func (c *Coordinator) DiagnosticHooks() diagnostics.Hooks {
 				return o
 			},
 		},
-		diagnostics.Hook{
-			Name:        "otel-merged",
-			Filename:    "otel-merged.yaml",
-			Description: "Final otel configuration used by the Elastic Agent. Includes hybrid mode config and component config.",
-			ContentType: "application/yaml",
-			Hook: func(_ context.Context) []byte {
-				mergedCfg := c.otelMgr.MergedOtelConfig()
-				if mergedCfg == nil {
-					return []byte("no active OTel configuration")
-				}
-				o, err := yaml.Marshal(mergedCfg.ToStringMap())
-				if err != nil {
-					return []byte(fmt.Sprintf("error: failed to convert to yaml: %v", err))
-				}
-				return o
-			},
-		},
 	}
 
-	hooks = append(hooks, c.otelMgr.DiagnosticHooks()...)
-
 	return hooks
 }
 

internal/pkg/agent/application/paths/common.go

@@ -58,17 +58,18 @@ const (
 var ExternalInputsPattern = filepath.Join("inputs.d", "*.yml")
 
 var (
-	topPath           string
-	configPath        string
-	configFilePath    string
-	logsPath          string
-	downloadsPath     string
-	componentsPath    string
-	installPath       string
-	controlSocketPath string
-	edotSocketPath    string
-	unversionedHome   bool
-	tmpCreator        sync.Once
+	topPath                    string
+	configPath                 string
+	configFilePath             string
+	logsPath                   string
+	downloadsPath              string
+	componentsPath             string
+	installPath                string
+	controlSocketPath          string
+	edotSocketPath             string
+	diagnosticsExtensionSocket string
+	unversionedHome            bool
+	tmpCreator                 sync.Once
 )
 
 func init() {
@@ -77,7 +78,8 @@ func init() {
 	configPath = topPath
 	logsPath = topPath
 	controlSocketPath = initialControlSocketPath(topPath)
-	edotSocketPath = EDOTSocketFromPath(runtime.GOOS, topPath)
+	edotSocketPath = SocketFromPath(runtime.GOOS, topPath, EDOTSocketName)
+	diagnosticsExtensionSocket = SocketFromPath(runtime.GOOS, topPath, DiagnosticsExtensionSocketName)
 	unversionedHome = false // only versioned by container subcommand
 
 	// these should never change
@@ -93,6 +95,7 @@ func init() {
 	fs.StringVar(&configFilePath, "c", DefaultConfigName, "Configuration file, relative to path.config")
 	fs.StringVar(&logsPath, "path.logs", logsPath, "Logs path contains Agent log output")
 	fs.StringVar(&controlSocketPath, "path.socket", controlSocketPath, "Control protocol socket path for the Agent")
+	fs.StringVar(&edotSocketPath, "path.edot_socket", edotSocketPath, "Control protocol socket path for the EDOT")
 
 	// enable user to download update artifacts to alternative place
 	// TODO: remove path.downloads support on next major (this can be configured using `agent.download.targetDirectory`)
@@ -352,27 +355,14 @@ func RunningInstalled() bool {
 // ControlSocketFromPath returns the control socket path for an Elastic Agent running
 // on the defined platform, and its executing directory.
 func ControlSocketFromPath(platform string, path string) string {
-	// socket should be inside this directory
-	socketPath := filepath.Join(path, ControlSocketName)
-	if platform == "windows" {
-		// on windows the control socket always uses the fallback
-		return utils.SocketURLWithFallback(socketPath, path)
-	}
-	unixSocket := fmt.Sprintf("unix://%s", socketPath)
-	if len(unixSocket) < 104 {
-		// small enough to fit
-		return unixSocket
-	}
-	// place in global /tmp to ensure that its small enough to fit; current path is way to long
-	// for it to be used, but needs to be unique per Agent (in the case that multiple are running)
-	return utils.SocketURLWithFallback(socketPath, path)
+	return SocketFromPath(platform, path, ControlSocketName)
 }
 
-// EDOTSocketFromPath returns the EDOT socket path for an Elastic Agent running
-// on the defined platform, and its executing directory.
-func EDOTSocketFromPath(platform string, path string) string {
+// SocketFromPath returns the socket path for an Elastic Agent running
+// on the defined platform for a given socket, and its executing directory.
+func SocketFromPath(platform string, path string, socketName string) string {
 	// socket should be inside this directory
-	socketPath := filepath.Join(path, EDOTSocketName)
+	socketPath := filepath.Join(path, socketName)
 	if platform == "windows" {
 		// on windows the control socket always uses the fallback
 		return utils.SocketURLWithFallback(socketPath, path)
@@ -388,7 +378,11 @@ func EDOTSocketFromPath(platform string, path string) string {
 }
 
 func DiagnosticsExtensionSocket() string {
-	return filepath.Join(topPath, DiagnosticsExtensionSocketName)
+	return diagnosticsExtensionSocket
+}
+
+func SetDiagnosticsExtensionSocket(socket string) {
+	diagnosticsExtensionSocket = SocketFromPath(runtime.GOOS, topPath, socket)
 }
 
 func pathSplit(path string) []string {

internal/pkg/agent/cmd/run.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/enroll"
 	fleetgateway "github.com/elastic/elastic-agent/internal/pkg/agent/application/gateway/fleet"
+	"github.com/elastic/elastic-agent/internal/pkg/diagnostics"
 
 	"go.elastic.co/apm/v2"
 	apmtransport "go.elastic.co/apm/v2/transport"
@@ -50,7 +51,6 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/cli"
 	"github.com/elastic/elastic-agent/internal/pkg/config"
 	monitoringCfg "github.com/elastic/elastic-agent/internal/pkg/core/monitoring/config"
-	"github.com/elastic/elastic-agent/internal/pkg/diagnostics"
 	"github.com/elastic/elastic-agent/internal/pkg/release"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/control/v2/server"
@@ -312,8 +312,10 @@ func runElasticAgent(
 		}
 	}()
 
+	diagHooks := diagnostics.GlobalHooks()
+	diagHooks = append(diagHooks, coord.DiagnosticHooks()...)
 	controlLog := l.Named("control")
-	control := server.New(controlLog, agentInfo, coord, tracer, diagnostics.GlobalHooks(), cfg.Settings.GRPC)
+	control := server.New(controlLog, agentInfo, coord, tracer, diagHooks, cfg.Settings.GRPC)
 
 	// if the configMgr implements the TestModeConfigSetter in means that Elastic Agent is in testing mode and
 	// the configuration will come in over the control protocol, so we set the config setting on the control protocol

internal/pkg/otel/manager/diagnostics.go

@@ -14,22 +14,18 @@ import (
 	"fmt"
 	"io"
 	"io/fs"
-	"net"
-	"net/http"
 	"os"
 	"path/filepath"
 	"regexp"
 	"strings"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring"
-	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
-	"github.com/elastic/elastic-agent/internal/pkg/diagnostics"
+	"github.com/elastic/elastic-agent/internal/pkg/otel"
 
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
 
-	"github.com/elastic/elastic-agent/internal/pkg/otel/elasticdiagnosticsextension"
 	"github.com/elastic/elastic-agent/internal/pkg/otel/translate"
 	"github.com/elastic/elastic-agent/pkg/core/logger"
 
@@ -40,26 +36,6 @@ import (
 
 var fileBeatRegistryPathRegExps = getRegexpsForRegistryFiles()
 
-func (m *OTelManager) DiagnosticHooks() (hooks diagnostics.Hooks) {
-	extDiagnostics, err := m.performDiagnosticsExt()
-	if err != nil {
-		m.logger.Errorf("error fetchign diagnostics: %v", err)
-		return
-	}
-	for _, hook := range extDiagnostics.GlobalDiagnostics {
-		hooks = append(hooks, diagnostics.Hook{
-			Name:        hook.Name,
-			Filename:    hook.Filename,
-			Description: hook.Description,
-			ContentType: hook.ContentType,
-			Hook: func(ctx context.Context) []byte {
-				return hook.Content
-			},
-		})
-	}
-	return
-}
-
 // PerformDiagnostics executes the diagnostic action for the provided units. If no units are provided then
 // it performs diagnostics for all current units. If a given unit does not exist in the manager, then a warning
 // is logged.
@@ -185,7 +161,7 @@ func (m *OTelManager) PerformComponentDiagnostics(
 		diagnostics[idx].Err = errors.Join(errs...)
 	}
 
-	extDiagnostics, err := m.performDiagnosticsExt()
+	extDiagnostics, err := otel.PerformDiagnosticsExt()
 	if err != nil {
 		m.logger.Errorf("error fetchign diagnostics: %v", err)
 		return nil, err
@@ -435,29 +411,3 @@ func matchRegistryFiles(registryFileRegExps []*regexp.Regexp, path string) bool
 	}
 	return false
 }
-
-func (m *OTelManager) performDiagnosticsExt() (*elasticdiagnosticsextension.Response, error) {
-	tr := &http.Transport{
-		DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
-			return net.Dial("unix", paths.DiagnosticsExtensionSocket())
-		},
-	}
-	client := &http.Client{Transport: tr}
-	resp, err := client.Get("http://localhost/diagnostics")
-	if err != nil {
-		return nil, err
-	}
-	defer resp.Body.Close()
-	respBytes, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return nil, err
-	}
-
-	var respSerialized elasticdiagnosticsextension.Response
-
-	if err := json.Unmarshal(respBytes, &respSerialized); err != nil {
-		return nil, err
-	}
-
-	return &respSerialized, nil
-}

internal/pkg/otel/manager/manager.go

@@ -339,12 +339,16 @@ func (m *OTelManager) buildMergedConfig(cfgUpdate configUpdate) (*confmap.Conf,
 	if err := m.injectDiagnosticsExtension(mergedOtelCfg); err != nil {
 		return nil, fmt.Errorf("failed to inject diagnostics: %w", err)
 	}
-	m.logger.Warnf("Here's complete config: %v", mergedOtelCfg.ToStringMap())
 
 	return mergedOtelCfg, nil
 }
 
 func (m *OTelManager) injectDiagnosticsExtension(config *confmap.Conf) error {
+	if _, err := os.Stat(paths.DiagnosticsExtensionSocket()); !os.IsNotExist(err) {
+		// socket already exists. User is most likely using "development" namespace.
+		// generate a unique socket name by fetching current timestamp
+		paths.SetDiagnosticsExtensionSocket(fmt.Sprintf("%s-edot-diagnostics-extension.sock", time.Now().UnixMilli()))
+	}
 	extensionCfg := map[string]any{
 		"extensions": map[string]any{
 			"elastic_diagnsotics": map[string]any{

internal/pkg/otel/server.go

@@ -62,7 +62,7 @@ func (s *server) Stop() error {
 }
 
 func (s *server) DiagnosticAgent(ctx context.Context, _ *cproto.DiagnosticAgentRequest) (*cproto.DiagnosticAgentResponse, error) {
-	resp, err := s.performDiagnosticsExt()
+	resp, err := PerformDiagnosticsExt()
 	if err != nil {
 		return nil, err
 	}
@@ -82,7 +82,7 @@ func (s *server) DiagnosticAgent(ctx context.Context, _ *cproto.DiagnosticAgentR
 }
 
 func (s *server) DiagnosticComponents(req *cproto.DiagnosticComponentsRequest, respServ cproto.ElasticAgentControl_DiagnosticComponentsServer) error {
-	resp, err := s.performDiagnosticsExt()
+	resp, err := PerformDiagnosticsExt()
 	if err != nil {
 		return err
 	}
@@ -106,7 +106,7 @@ func (s *server) DiagnosticComponents(req *cproto.DiagnosticComponentsRequest, r
 	return nil
 }
 
-func (s *server) performDiagnosticsExt() (*elasticdiagnosticsextension.Response, error) {
+func PerformDiagnosticsExt() (*elasticdiagnosticsextension.Response, error) {
 	tr := &http.Transport{
 		DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
 			return net.Dial("unix", paths.DiagnosticsExtensionSocket())

pkg/control/v2/server/server.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"syscall"
 	"time"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status"
@@ -31,6 +32,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/configuration"
 	"github.com/elastic/elastic-agent/internal/pkg/diagnostics"
+	"github.com/elastic/elastic-agent/internal/pkg/otel"
 	"github.com/elastic/elastic-agent/internal/pkg/release"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/component/runtime"
@@ -194,12 +196,8 @@ func (s *Server) Upgrade(ctx context.Context, request *cproto.UpgradeRequest) (*
 
 // DiagnosticAgent returns diagnostic information for this running Elastic Agent.
 func (s *Server) DiagnosticAgent(ctx context.Context, req *cproto.DiagnosticAgentRequest) (*cproto.DiagnosticAgentResponse, error) {
-
-	// DiagnosticHooks() is called on every DiagnosticAgent request,
-	// since otelMgr retrieves the hook list from diagnosticsExtension.
-	diagHooks := append(s.diagHooks, s.coord.DiagnosticHooks()...)
-	res := make([]*cproto.DiagnosticFileResult, 0, len(diagHooks))
-	for _, h := range diagHooks {
+	res := make([]*cproto.DiagnosticFileResult, 0, len(s.diagHooks))
+	for _, h := range s.diagHooks {
 		if ctx.Err() != nil {
 			return nil, ctx.Err()
 		}
@@ -234,6 +232,27 @@ func (s *Server) DiagnosticAgent(ctx context.Context, req *cproto.DiagnosticAgen
 		}
 	}
 
+	resp, err := otel.PerformDiagnosticsExt()
+	if errors.Is(err, syscall.ENOENT) || errors.Is(err, syscall.ECONNREFUSED) {
+		// We're not running the EDOT if:
+		//  1. Either the socket doesn't exist
+		//	2. It is refusing the connections.
+		s.logger.Debugf("Couldn't fetch diagnostics from EDOT: %v", err)
+		return &cproto.DiagnosticAgentResponse{Results: res}, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	for _, r := range resp.GlobalDiagnostics {
+		res = append(res, &cproto.DiagnosticFileResult{
+			Name:        r.Name,
+			Filename:    r.Filename,
+			ContentType: r.ContentType,
+			Content:     r.Content,
+			Description: r.Description,
+		})
+	}
+
 	if ctx.Err() != nil {
 		return nil, ctx.Err()
 	}