Revert "Remove jaeger-agent from distributions (#6081)"

This reverts commit ac6f78b.

Makefile.BuildBinaries.mk

@@ -81,6 +81,10 @@ build-jaeger: build-ui _build-a-binary-jaeger$(SUFFIX)-$(GOOS)-$(GOARCH)
 build-all-in-one: BIN_NAME = all-in-one
 build-all-in-one: build-ui _build-a-binary-all-in-one$(SUFFIX)-$(GOOS)-$(GOARCH)
 
+.PHONY: build-agent
+build-agent: BIN_NAME = agent
+build-agent: _build-a-binary-agent$(SUFFIX)-$(GOOS)-$(GOARCH)
+
 .PHONY: build-query
 build-query: BIN_NAME = query
 build-query: build-ui _build-a-binary-query$(SUFFIX)-$(GOOS)-$(GOARCH)
@@ -131,6 +135,7 @@ build-binaries-linux-ppc64le:
 # build all binaries for one specific platform GOOS/GOARCH
 .PHONY: _build-platform-binaries
 _build-platform-binaries: \
+		build-agent \
 		build-all-in-one \
 		build-collector \
 		build-query \
@@ -150,6 +155,7 @@ _build-platform-binaries: \
 .PHONY: _build-platform-binaries-debug
 _build-platform-binaries-debug:
 _build-platform-binaries-debug: \
+	build-agent \
 	build-collector \
 	build-query \
 	build-ingester \

Makefile.Windows.mk

@@ -64,6 +64,7 @@ _build-syso: $(GOVERSIONINFO)
 	$(eval SEMVER_MAJOR := $(word 2, $(SEMVER_ALL)))
 	$(eval SEMVER_MINOR := $(word 3, $(SEMVER_ALL)))
 	$(eval SEMVER_PATCH := $(word 4, $(SEMVER_ALL)))
+	$(call _build_syso_macro,Jaeger Agent,cmd/agent)
 	$(call _build_syso_macro,Jaeger Collector,cmd/collector)
 	$(call _build_syso_macro,Jaeger Query,cmd/query)
 	$(call _build_syso_macro,Jaeger Ingester,cmd/ingester)

cmd/agent/app/builder.go

@@ -22,6 +22,7 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/agent/app/servers/thriftudp"
 	"github.com/jaegertracing/jaeger/internal/safeexpvar"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
+	"github.com/jaegertracing/jaeger/ports"
 	agentThrift "github.com/jaegertracing/jaeger/thrift-gen/agent"
 )
 
@@ -37,7 +38,7 @@ const (
 	binaryProtocol  Protocol = "binary"
 )
 
-var defaultHTTPServerHostPort = ":" + strconv.Itoa(AgentConfigServerHTTP)
+var defaultHTTPServerHostPort = ":" + strconv.Itoa(ports.AgentConfigServerHTTP)
 
 // Model used to distinguish the data transfer model
 type Model string

cmd/agent/app/flags.go

@@ -10,6 +10,8 @@ import (
 	"strconv"
 
 	"github.com/spf13/viper"
+
+	"github.com/jaegertracing/jaeger/ports"
 )
 
 const (
@@ -28,9 +30,9 @@ var defaultProcessors = []struct {
 	protocol Protocol
 	port     int
 }{
-	{model: "zipkin", protocol: "compact", port: AgentZipkinThriftCompactUDP},
-	{model: "jaeger", protocol: "compact", port: AgentJaegerThriftCompactUDP},
-	{model: "jaeger", protocol: "binary", port: AgentJaegerThriftBinaryUDP},
+	{model: "zipkin", protocol: "compact", port: ports.AgentZipkinThriftCompactUDP},
+	{model: "jaeger", protocol: "compact", port: ports.AgentJaegerThriftCompactUDP},
+	{model: "jaeger", protocol: "binary", port: ports.AgentJaegerThriftBinaryUDP},
 }
 
 // AddFlags adds flags for Builder.

cmd/agent/main.go

@@ -23,14 +23,15 @@ import (
 	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
 	"github.com/jaegertracing/jaeger/pkg/version"
+	"github.com/jaegertracing/jaeger/ports"
 )
 
 func main() {
 	println("***************************************************************************************************")
 	println("*** WARNING jaeger-agent is deprecated. See https://github.com/jaegertracing/jaeger/issues/4739 ***")
 	println("***************************************************************************************************")
 
-	svc := flags.NewService(app.AgentAdminHTTP)
+	svc := flags.NewService(ports.AgentAdminHTTP)
 	svc.NoStorage = true
 
 	v := viper.New()
@@ -91,7 +92,7 @@ func main() {
 
 	command.AddCommand(version.Command())
 	command.AddCommand(docs.Command(v))
-	command.AddCommand(status.Command(v, app.AgentAdminHTTP))
+	command.AddCommand(status.Command(v, ports.AgentAdminHTTP))
 
 	config.AddFlags(
 		v,

cmd/all-in-one/Dockerfile

@@ -8,6 +8,18 @@ FROM $base_image AS release
 ARG TARGETARCH
 ARG USER_UID=10001
 
+# Agent zipkin.thrift compact
+EXPOSE 5775/udp
+
+# Agent jaeger.thrift compact
+EXPOSE 6831/udp
+
+# Agent jaeger.thrift binary
+EXPOSE 6832/udp
+
+# Agent config HTTP
+EXPOSE 5778
+
 # Collector OTLP gRPC
 EXPOSE 4317
 
@@ -40,6 +52,18 @@ FROM $debug_image AS debug
 ARG TARGETARCH=amd64
 ARG USER_UID=10001
 
+# Agent zipkin.thrift compact
+EXPOSE 5775/udp
+
+# Agent jaeger.thrift compact
+EXPOSE 6831/udp
+
+# Agent jaeger.thrift binary
+EXPOSE 6832/udp
+
+# Agent config HTTP
+EXPOSE 5778
+
 # Collector OTLP gRPC
 EXPOSE 4317
 

cmd/all-in-one/all_in_one_test.go

@@ -7,7 +7,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"fmt"
 	"io"
 	"net/http"
 	"os"
@@ -21,25 +20,24 @@ import (
 	"github.com/stretchr/testify/require"
 
 	ui "github.com/jaegertracing/jaeger/model/json"
-	"github.com/jaegertracing/jaeger/ports"
 	"github.com/jaegertracing/jaeger/proto-gen/api_v2"
 )
 
 // These tests are only run when the environment variable TEST_MODE=integration is set.
 
 const (
-	host = "0.0.0.0"
+	host       = "0.0.0.0"
+	queryPort  = "16686"
+	agentPort  = "5778"
+	healthPort = "13133"
+	queryAddr  = "http://" + host + ":" + queryPort
+	agentAddr  = "http://" + host + ":" + agentPort
+	healthAddr = "http://" + host + ":" + healthPort + "/status"
 
 	getServicesURL         = "/api/services"
 	getTraceURL            = "/api/traces/"
 	getServicesAPIV3URL    = "/api/v3/services"
-	getSamplingStrategyURL = "/api/sampling?service=whatever"
-)
-
-var (
-	queryAddr    = fmt.Sprintf("http://%s:%d", host, ports.QueryHTTP)
-	samplingAddr = fmt.Sprintf("http://%s:%d", host, ports.CollectorHTTP)
-	healthAddr   = fmt.Sprintf("http://%s:%d/status", host, ports.CollectorV2HealthChecks)
+	getSamplingStrategyURL = "/sampling?service=whatever"
 )
 
 var traceID string // stores state exchanged between createTrace and getAPITrace
@@ -172,7 +170,7 @@ func getAPITrace(t *testing.T) {
 func getSamplingStrategy(t *testing.T) {
 	// TODO should we test refreshing the strategy file?
 
-	r, body := httpGet(t, samplingAddr+getSamplingStrategyURL)
+	r, body := httpGet(t, agentAddr+getSamplingStrategyURL)
 	t.Logf("Sampling strategy response: %s", string(body))
 	require.EqualValues(t, http.StatusOK, r.StatusCode)
 

cmd/all-in-one/main.go

@@ -6,7 +6,6 @@ package main
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -17,6 +16,9 @@ import (
 	_ "go.uber.org/automaxprocs"
 	"go.uber.org/zap"
 
+	agentApp "github.com/jaegertracing/jaeger/cmd/agent/app"
+	agentRep "github.com/jaegertracing/jaeger/cmd/agent/app/reporter"
+	agentGrpcRep "github.com/jaegertracing/jaeger/cmd/agent/app/reporter/grpc"
 	"github.com/jaegertracing/jaeger/cmd/all-in-one/setupcontext"
 	collectorApp "github.com/jaegertracing/jaeger/cmd/collector/app"
 	collectorFlags "github.com/jaegertracing/jaeger/cmd/collector/app/flags"
@@ -74,8 +76,8 @@ func main() {
 	v := viper.New()
 	command := &cobra.Command{
 		Use:   "jaeger-all-in-one",
-		Short: "Jaeger all-in-one distribution with collector and query in one process.",
-		Long: `Jaeger all-in-one distribution with collector and query. Use with caution: this version
+		Short: "Jaeger all-in-one distribution with agent, collector and query in one process.",
+		Long: `Jaeger all-in-one distribution with agent, collector and query. Use with caution this version
 by default uses only in-memory database.`,
 		RunE: func(_ *cobra.Command, _ /* args */ []string) error {
 			if err := svc.Start(v); err != nil {
@@ -84,6 +86,7 @@ by default uses only in-memory database.`,
 			logger := svc.Logger // shortcut
 			baseFactory := svc.MetricsFactory.Namespace(metrics.NSOptions{Name: "jaeger"})
 			version.NewInfoMetrics(baseFactory)
+			agentMetricsFactory := baseFactory.Namespace(metrics.NSOptions{Name: "agent"})
 			collectorMetricsFactory := baseFactory.Namespace(metrics.NSOptions{Name: "collector"})
 			queryMetricsFactory := baseFactory.Namespace(metrics.NSOptions{Name: "query"})
 
@@ -129,6 +132,12 @@ by default uses only in-memory database.`,
 				logger.Fatal("Failed to create sampling strategy provider", zap.Error(err))
 			}
 
+			aOpts := new(agentApp.Builder).InitFromViper(v)
+			repOpts := new(agentRep.Options).InitFromViper(v, logger)
+			grpcBuilder, err := agentGrpcRep.NewConnBuilder().InitFromViper(v)
+			if err != nil {
+				logger.Fatal("Failed to configure connection for grpc", zap.Error(err))
+			}
 			cOpts, err := new(collectorFlags.CollectorOptions).InitFromViper(v, logger)
 			if err != nil {
 				logger.Fatal("Failed to initialize collector", zap.Error(err))
@@ -155,6 +164,25 @@ by default uses only in-memory database.`,
 				log.Fatal(err)
 			}
 
+			// agent
+			// if the agent reporter grpc host:port was not explicitly set then use whatever the collector is listening on
+			if len(grpcBuilder.CollectorHostPorts) == 0 {
+				grpcBuilder.CollectorHostPorts = append(grpcBuilder.CollectorHostPorts, cOpts.GRPC.HostPort)
+			}
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			builders := map[agentRep.Type]agentApp.CollectorProxyBuilder{
+				agentRep.GRPC: agentApp.GRPCCollectorProxyBuilder(grpcBuilder),
+			}
+			cp, err := agentApp.CreateCollectorProxy(ctx, agentApp.ProxyBuilderOptions{
+				Options: *repOpts,
+				Logger:  logger,
+				Metrics: agentMetricsFactory,
+			}, builders)
+			if err != nil {
+				logger.Fatal("Could not create collector proxy", zap.Error(err))
+			}
+			agent := startAgent(cp, aOpts, logger, agentMetricsFactory)
 			telset := telemetery.Setting{
 				Logger:         svc.Logger,
 				TracerProvider: tracer.OTEL,
@@ -169,16 +197,20 @@ by default uses only in-memory database.`,
 			)
 
 			svc.RunAndThen(func() {
-				var errs []error
-				errs = append(errs, c.Close())
-				errs = append(errs, querySrv.Close())
+				agent.Stop()
+				_ = cp.Close()
+				_ = c.Close()
+				_ = querySrv.Close()
 				if closer, ok := spanWriter.(io.Closer); ok {
-					errs = append(errs, closer.Close())
+					if err := closer.Close(); err != nil {
+						logger.Error("Failed to close span writer", zap.Error(err))
+					}
 				}
-				errs = append(errs, storageFactory.Close())
-				errs = append(errs, tracer.Close(context.Background()))
-				if err := errors.Join(errs...); err != nil {
-					logger.Error("Failed to close services", zap.Error(err))
+				if err := storageFactory.Close(); err != nil {
+					logger.Error("Failed to close storage factory", zap.Error(err))
+				}
+				if err := tracer.Close(context.Background()); err != nil {
+					logger.Error("Error shutting down tracer provider", zap.Error(err))
 				}
 			})
 			return nil
@@ -196,6 +228,9 @@ by default uses only in-memory database.`,
 		command,
 		svc.AddFlags,
 		storageFactory.AddPipelineFlags,
+		agentApp.AddFlags,
+		agentRep.AddFlags,
+		agentGrpcRep.AddFlags,
 		collectorFlags.AddFlags,
 		queryApp.AddFlags,
 		samplingStrategyFactory.AddFlags,
@@ -207,6 +242,25 @@ by default uses only in-memory database.`,
 	}
 }
 
+func startAgent(
+	cp agentApp.CollectorProxy,
+	b *agentApp.Builder,
+	logger *zap.Logger,
+	baseFactory metrics.Factory,
+) *agentApp.Agent {
+	agent, err := b.CreateAgent(cp, logger, baseFactory)
+	if err != nil {
+		logger.Fatal("Unable to initialize Jaeger Agent", zap.Error(err))
+	}
+
+	logger.Info("Starting agent")
+	if err := agent.Run(); err != nil {
+		logger.Fatal("Failed to run the agent", zap.Error(err))
+	}
+
+	return agent
+}
+
 func startQuery(
 	svc *flags.Service,
 	qOpts *queryApp.QueryOptions,

cmd/collector/main.go

@@ -52,8 +52,8 @@ func main() {
 	v := viper.New()
 	command := &cobra.Command{
 		Use:   "jaeger-collector",
-		Short: "Jaeger collector receives and stores traces",
-		Long:  `Jaeger collector receives traces and runs them through a processing pipeline.`,
+		Short: "Jaeger collector receives and processes traces from Jaeger agents and clients",
+		Long:  `Jaeger collector receives traces from Jaeger agents and runs them through a processing pipeline.`,
 		RunE: func(_ *cobra.Command, _ /* args */ []string) error {
 			if err := svc.Start(v); err != nil {
 				return err

cmd/jaeger/Dockerfile

@@ -10,6 +10,15 @@ FROM $base_image AS release
 ARG TARGETARCH
 ARG USER_UID=10001
 
+# Agent zipkin.thrift compact
+EXPOSE 5775/udp
+
+# Agent jaeger.thrift compact
+EXPOSE 6831/udp
+
+# Agent jaeger.thrift binary
+EXPOSE 6832/udp
+
 # Sampling config HTTP
 EXPOSE 5778
 
@@ -53,6 +62,15 @@ FROM $debug_image AS debug
 ARG TARGETARCH=amd64
 ARG USER_UID=10001
 
+# Agent zipkin.thrift compact
+EXPOSE 5775/udp
+
+# Agent jaeger.thrift compact
+EXPOSE 6831/udp
+
+# Agent jaeger.thrift binary
+EXPOSE 6832/udp
+
 # Sampling config HTTP
 EXPOSE 5778