feat: remove references to es output and unused callbacks (#19554)

APM Server no longer uses the beats elasticsearch output
to send data to ES.
Remove unused callbacks and references

Author: kruskall

internal/beatcmd/beat.go

@@ -46,12 +46,9 @@ import (
 	"github.com/elastic/beats/v7/libbeat/api"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common/reload"
-	"github.com/elastic/beats/v7/libbeat/esleg/eslegclient"
-	"github.com/elastic/beats/v7/libbeat/licenser"
 	"github.com/elastic/beats/v7/libbeat/management"
 	"github.com/elastic/beats/v7/libbeat/monitoring/report"
 	"github.com/elastic/beats/v7/libbeat/monitoring/report/log"
-	"github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
 	"github.com/elastic/beats/v7/libbeat/pprof"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/file"
@@ -61,7 +58,6 @@ import (
 	"github.com/elastic/elastic-agent-libs/monitoring/report/buffer"
 	"github.com/elastic/elastic-agent-libs/paths"
 	"github.com/elastic/elastic-agent-libs/service"
-	libversion "github.com/elastic/elastic-agent-libs/version"
 	"github.com/elastic/elastic-agent-system-metrics/metric/system/host"
 	metricreport "github.com/elastic/elastic-agent-system-metrics/report"
 	sysinfo "github.com/elastic/go-sysinfo"
@@ -379,18 +375,6 @@ func (b *Beat) Run(ctx context.Context) error {
 
 	logSystemInfo(b.Info)
 
-	cleanup, err := b.registerElasticsearchVersionCheck()
-	if err != nil {
-		return err
-	}
-	defer cleanup()
-
-	cleanup, err = b.registerClusterUUIDFetching()
-	if err != nil {
-		return err
-	}
-	defer cleanup()
-
 	statsRegistry := b.Monitoring.StatsRegistry()
 
 	if err := metricreport.SetupMetricsOptions(metricreport.MetricOptions{
@@ -739,75 +723,6 @@ func addDocappenderOutputElasticsearchMetrics(ctx context.Context, v monitoring.
 	v.OnRegistryFinished()
 }
 
-// registerElasticsearchVerfication registers a global callback to make sure
-// the Elasticsearch instance we are connecting to has a valid license, and is
-// at least on the same version as APM Server.
-//
-// registerElasticsearchVerification returns a cleanup function which must be
-// called on shutdown.
-func (b *Beat) registerElasticsearchVersionCheck() (func(), error) {
-	uuid, err := elasticsearch.RegisterGlobalCallback(func(conn *eslegclient.Connection, logger *logp.Logger) error {
-		if err := licenser.FetchAndVerify(conn, logger); err != nil {
-			return err
-		}
-		esVersion := conn.GetVersion()
-		beatVersion, err := libversion.New(b.Info.Version)
-		if err != nil {
-			return err
-		}
-		if esVersion.LessThanMajorMinor(beatVersion) {
-			return fmt.Errorf(
-				"%w Elasticsearch: %s, APM Server: %s",
-				elasticsearch.ErrTooOld, esVersion.String(), b.Info.Version,
-			)
-		}
-		return nil
-	})
-	if err != nil {
-		return nil, err
-	}
-	return func() { elasticsearch.DeregisterGlobalCallback(uuid) }, nil
-}
-
-func (b *Beat) registerClusterUUIDFetching() (func(), error) {
-	callback := b.clusterUUIDFetchingCallback()
-	uuid, err := elasticsearch.RegisterConnectCallback(callback)
-	if err != nil {
-		return nil, err
-	}
-	return func() { elasticsearch.DeregisterConnectCallback(uuid) }, nil
-}
-
-// Build and return a callback to fetch the Elasticsearch cluster_uuid for monitoring
-func (b *Beat) clusterUUIDFetchingCallback() elasticsearch.ConnectCallback {
-	stateRegistry := b.Monitoring.StateRegistry()
-	elasticsearchRegistry := stateRegistry.GetOrCreateRegistry("outputs.elasticsearch")
-	clusterUUIDRegVar := monitoring.NewString(elasticsearchRegistry, "cluster_uuid")
-
-	callback := func(esClient *eslegclient.Connection, _ *logp.Logger) error {
-		var response struct {
-			ClusterUUID string `json:"cluster_uuid"`
-		}
-
-		status, body, err := esClient.Request("GET", "/", "", nil, nil)
-		if err != nil {
-			return fmt.Errorf("error querying /: %w", err)
-		}
-		if status > 299 {
-			return fmt.Errorf("error querying /. Status: %d. Response body: %s", status, body)
-		}
-		err = json.Unmarshal(body, &response)
-		if err != nil {
-			return fmt.Errorf("error unmarshaling json when querying /. Body: %s", body)
-		}
-
-		clusterUUIDRegVar.Set(response.ClusterUUID)
-		return nil
-	}
-
-	return callback
-}
-
 func (b *Beat) setupMonitoring() (report.Reporter, error) {
 	monitoringCfg := b.Config.Monitoring
 

internal/beater/beater.go

@@ -46,11 +46,9 @@ import (
 	_ "google.golang.org/grpc/encoding/gzip"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
-	"github.com/elastic/beats/v7/libbeat/esleg/eslegclient"
 	"github.com/elastic/beats/v7/libbeat/instrumentation"
 	"github.com/elastic/beats/v7/libbeat/licenser"
 	"github.com/elastic/beats/v7/libbeat/outputs"
-	esoutput "github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
 	"github.com/elastic/beats/v7/libbeat/publisher/pipeline"
 	"github.com/elastic/beats/v7/libbeat/publisher/pipetool"
 	agentconfig "github.com/elastic/elastic-agent-libs/config"
@@ -318,18 +316,6 @@ func (s *Runner) Run(ctx context.Context) error {
 		close(publishReady)
 		return nil
 	})
-	callbackUUID, err := esoutput.RegisterConnectCallback(func(*eslegclient.Connection, *logp.Logger) error {
-		select {
-		case <-publishReady:
-			return nil
-		default:
-		}
-		return errors.New("not ready for publishing events")
-	})
-	if err != nil {
-		return err
-	}
-	defer esoutput.DeregisterConnectCallback(callbackUUID)
 	newESClient := func(tp trace.TracerProvider) func(cfg *elasticsearch.Config, logger *logp.Logger) (*elasticsearch.Client, error) {
 		return func(cfg *elasticsearch.Config, logger *logp.Logger) (*elasticsearch.Client, error) {
 			httpTransport, err := elasticsearch.NewHTTPTransport(cfg, logger)

internal/publish/pub_test.go

@@ -36,7 +36,6 @@ import (
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/idxmgmt"
 	"github.com/elastic/beats/v7/libbeat/outputs"
-	_ "github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
 	"github.com/elastic/beats/v7/libbeat/publisher"
 	"github.com/elastic/beats/v7/libbeat/publisher/pipeline"
 	"github.com/elastic/beats/v7/libbeat/publisher/pipetool"